From e8a291de7885e311f584af769a308cc1e58cb687 Mon Sep 17 00:00:00 2001
From: Benedikt Labrenz <benedikt@labrenz.org>
Date: Thu, 1 Oct 2026 13:58:37 +0200
Subject: [PATCH 1/9] add agent label utility

---
 crates/stackable-operator/src/v2/kvp/label.rs | 80 +++++++++++++++++++
 1 file changed, 80 insertions(+)

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(

From 6ca45a0656f0a623061a3bfbff52c5ae113c352e Mon Sep 17 00:00:00 2001
From: Benedikt Labrenz <benedikt@labrenz.org>
Date: Thu, 1 Oct 2026 14:03:46 +0200
Subject: [PATCH 2/9] add TlsClientCredential enum as platform access utility

---
 crates/stackable-operator/src/commons/mod.rs  |   1 +
 .../src/commons/platform_access/mod.rs        |   3 +
 .../src/commons/platform_access/tls.rs        | 135 ++++++++++++++++++
 3 files changed, 139 insertions(+)
 create mode 100644 crates/stackable-operator/src/commons/platform_access/mod.rs
 create mode 100644 crates/stackable-operator/src/commons/platform_access/tls.rs

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/mod.rs b/crates/stackable-operator/src/commons/platform_access/mod.rs
new file mode 100644
index 000000000..7cedc76bf
--- /dev/null
+++ b/crates/stackable-operator/src/commons/platform_access/mod.rs
@@ -0,0 +1,3 @@
+//! Credentials a product cluster grants the platform-access agent, and how to mount them.
+
+pub mod tls;
diff --git a/crates/stackable-operator/src/commons/platform_access/tls.rs b/crates/stackable-operator/src/commons/platform_access/tls.rs
new file mode 100644
index 000000000..9660e6ea7
--- /dev/null
+++ b/crates/stackable-operator/src/commons/platform_access/tls.rs
@@ -0,0 +1,135 @@
+use k8s_openapi::api::core::v1::{SecretVolumeSource, Volume, VolumeMount};
+use schemars::JsonSchema;
+use serde::{Deserialize, Serialize};
+
+use crate::{
+    builder::pod::{
+        PodBuilder,
+        container::ContainerBuilder,
+        volume::{
+            SecretFormat, SecretOperatorVolumeSourceBuilder, VolumeBuilder, VolumeMountBuilder,
+        },
+    },
+    commons::secret_class::SecretClassVolumeProvisionParts,
+    constants::secret::SECRET_BASE_PATH,
+    v2::types::kubernetes::{SecretClassName, SecretName},
+};
+
+/// 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),
+}
+
+impl TlsClientCredential {
+    /// Adds the certificate volume to the Pod and mounts it into all given containers.
+    /// - TlsClientCredential::Secret mounts the Secret
+    /// - TlsClientCredential::Secret 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("The volume name is derived from the credential and should not collide.");
+        for container_builder in container_builders {
+            container_builder
+                .add_volume_mounts(mounts.clone())
+                .expect("The mount path is derived from the credential and should not collide.");
+        }
+    }
+
+    fn volumes_and_mounts(&self) -> (Vec<Volume>, Vec<VolumeMount>) {
+        let volume_name = self.volume_name();
+        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) => Volume {
+                name: volume_name.clone(),
+                secret: Some(SecretVolumeSource {
+                    secret_name: Some(secret.to_string()),
+                    ..SecretVolumeSource::default()
+                }),
+                ..Volume::default()
+            },
+        };
+        let mount = VolumeMountBuilder::new(&volume_name, self.mount_path()).build();
+        (vec![volume], vec![mount])
+    }
+
+    /// The directory containing the certificate as `tls.crt` and `tls.key` (PEM).
+    pub fn mount_path(&self) -> String {
+        format!("{SECRET_BASE_PATH}/{}", self.volume_name())
+    }
+
+    fn volume_name(&self) -> String {
+        match self {
+            Self::SecretClass(secret_class) => format!("{secret_class}-tls-cert"),
+            Self::Secret(secret) => format!("{secret}-tls-cert"),
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use serde_json::json;
+
+    use super::*;
+
+    #[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-tls-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-tls-cert", "name": "tls-tls-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": "my-cert-tls-cert", "secret": {"secretName": "my-cert"}}])
+        );
+        assert_eq!(
+            serde_json::to_value(mounts).expect("serializable"),
+            json!([{"mountPath": "/stackable/secrets/my-cert-tls-cert", "name": "my-cert-tls-cert"}])
+        );
+    }
+}

From 7cb677e73bd218bcafb4ed4d4b459dda319fcbac Mon Sep 17 00:00:00 2001
From: Benedikt Labrenz <benedikt@labrenz.org>
Date: Thu, 1 Oct 2026 14:04:30 +0200
Subject: [PATCH 3/9] add changelog entry

---
 crates/stackable-operator/CHANGELOG.md | 8 ++++++++
 1 file changed, 8 insertions(+)

diff --git a/crates/stackable-operator/CHANGELOG.md b/crates/stackable-operator/CHANGELOG.md
index ee60b2114..6b9787862 100644
--- a/crates/stackable-operator/CHANGELOG.md
+++ b/crates/stackable-operator/CHANGELOG.md
@@ -4,6 +4,14 @@ 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.
+  - Labels and selectors for agents.
+
+[#1290]: https://github.com/stackabletech/operator-rs/pull/1290
+
 ## [0.119.0] - 2026-09-23
 
 ### Removed

From 7a0d956028f6382dee602ba370202d8b801b66e2 Mon Sep 17 00:00:00 2001
From: Benedikt Labrenz <benedikt@labrenz.org>
Date: Thu, 1 Oct 2026 14:12:45 +0200
Subject: [PATCH 4/9] fix cargo deny by upgrading yoke-derive

---
 Cargo.lock | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

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",

From f3d2da737aa49086cbdaba806aaf2c8645251cfb Mon Sep 17 00:00:00 2001
From: Benedikt Labrenz <benedikt@labrenz.org>
Date: Thu, 1 Oct 2026 16:37:52 +0200
Subject: [PATCH 5/9] use hard-coded volume and mount names

---
 .../src/commons/platform_access/tls.rs        | 48 ++++++++-----------
 1 file changed, 20 insertions(+), 28 deletions(-)

diff --git a/crates/stackable-operator/src/commons/platform_access/tls.rs b/crates/stackable-operator/src/commons/platform_access/tls.rs
index 9660e6ea7..3f383561b 100644
--- a/crates/stackable-operator/src/commons/platform_access/tls.rs
+++ b/crates/stackable-operator/src/commons/platform_access/tls.rs
@@ -1,3 +1,4 @@
+use const_format::concatcp;
 use k8s_openapi::api::core::v1::{SecretVolumeSource, Volume, VolumeMount};
 use schemars::JsonSchema;
 use serde::{Deserialize, Serialize};
@@ -15,6 +16,12 @@ use crate::{
     v2::types::kubernetes::{SecretClassName, SecretName},
 };
 
+/// 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);
+
 /// 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")]
@@ -30,7 +37,7 @@ pub enum TlsClientCredential {
 impl TlsClientCredential {
     /// Adds the certificate volume to the Pod and mounts it into all given containers.
     /// - TlsClientCredential::Secret mounts the Secret
-    /// - TlsClientCredential::Secret adds a secret-operator volume
+    /// - TlsClientCredential::SecretClass adds a secret-operator volume
     pub fn add_volumes_and_mounts(
         &self,
         pod_builder: &mut PodBuilder,
@@ -39,18 +46,17 @@ impl TlsClientCredential {
         let (volumes, mounts) = self.volumes_and_mounts();
         pod_builder
             .add_volumes(volumes)
-            .expect("The volume name is derived from the credential and should not collide.");
+            .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("The mount path is derived from the credential and should not collide.");
+                .expect("Only a single platform access authentication variant can be chosen.");
         }
     }
 
     fn volumes_and_mounts(&self) -> (Vec<Volume>, Vec<VolumeMount>) {
-        let volume_name = self.volume_name();
         let volume = match self {
-            Self::SecretClass(secret_class) => VolumeBuilder::new(&volume_name)
+            Self::SecretClass(secret_class) => VolumeBuilder::new(VOLUME_NAME)
                 .ephemeral(
                     SecretOperatorVolumeSourceBuilder::new(
                         secret_class,
@@ -62,30 +68,16 @@ impl TlsClientCredential {
                     .expect("the annotations are built from a valid SecretClass name"),
                 )
                 .build(),
-            Self::Secret(secret) => Volume {
-                name: volume_name.clone(),
-                secret: Some(SecretVolumeSource {
+            Self::Secret(secret) => VolumeBuilder::new(VOLUME_NAME)
+                .secret(SecretVolumeSource {
                     secret_name: Some(secret.to_string()),
                     ..SecretVolumeSource::default()
-                }),
-                ..Volume::default()
-            },
+                })
+                .build(),
         };
-        let mount = VolumeMountBuilder::new(&volume_name, self.mount_path()).build();
+        let mount = VolumeMountBuilder::new(VOLUME_NAME, MOUNT_PATH).build();
         (vec![volume], vec![mount])
     }
-
-    /// The directory containing the certificate as `tls.crt` and `tls.key` (PEM).
-    pub fn mount_path(&self) -> String {
-        format!("{SECRET_BASE_PATH}/{}", self.volume_name())
-    }
-
-    fn volume_name(&self) -> String {
-        match self {
-            Self::SecretClass(secret_class) => format!("{secret_class}-tls-cert"),
-            Self::Secret(secret) => format!("{secret}-tls-cert"),
-        }
-    }
 }
 
 #[cfg(test)]
@@ -101,7 +93,7 @@ mod tests {
         let (volumes, mounts) = credential.volumes_and_mounts();
         let volumes = serde_json::to_value(volumes).expect("serializable");
 
-        assert_eq!(volumes[0]["name"], "tls-tls-cert");
+        assert_eq!(volumes[0]["name"], "tls-client-cert");
         assert_eq!(
             volumes[0]["ephemeral"]["volumeClaimTemplate"]["metadata"]["annotations"],
             json!({
@@ -113,7 +105,7 @@ mod tests {
         );
         assert_eq!(
             serde_json::to_value(mounts).expect("serializable"),
-            json!([{"mountPath": "/stackable/secrets/tls-tls-cert", "name": "tls-tls-cert"}])
+            json!([{"mountPath": "/stackable/secrets/tls-client-cert", "name": "tls-client-cert"}])
         );
     }
 
@@ -125,11 +117,11 @@ mod tests {
 
         assert_eq!(
             serde_json::to_value(volumes).expect("serializable"),
-            json!([{"name": "my-cert-tls-cert", "secret": {"secretName": "my-cert"}}])
+            json!([{"name": "tls-client-cert", "secret": {"secretName": "my-cert"}}])
         );
         assert_eq!(
             serde_json::to_value(mounts).expect("serializable"),
-            json!([{"mountPath": "/stackable/secrets/my-cert-tls-cert", "name": "my-cert-tls-cert"}])
+            json!([{"mountPath": "/stackable/secrets/tls-client-cert", "name": "tls-client-cert"}])
         );
     }
 }

From 3b91d73a3e66639654597d6304780b5ccc75d9c0 Mon Sep 17 00:00:00 2001
From: Benedikt Labrenz <benedikt@labrenz.org>
Date: Tue, 6 Oct 2026 14:37:59 +0200
Subject: [PATCH 6/9] add support for anonymous platform access

---
 crates/stackable-operator/CHANGELOG.md        |  1 +
 .../src/commons/platform_access/anonymous.rs  | 27 +++++++++++++++++++
 .../src/commons/platform_access/mod.rs        |  3 ++-
 3 files changed, 30 insertions(+), 1 deletion(-)
 create mode 100644 crates/stackable-operator/src/commons/platform_access/anonymous.rs

diff --git a/crates/stackable-operator/CHANGELOG.md b/crates/stackable-operator/CHANGELOG.md
index 6b9787862..0d862e6fd 100644
--- a/crates/stackable-operator/CHANGELOG.md
+++ b/crates/stackable-operator/CHANGELOG.md
@@ -8,6 +8,7 @@ All notable changes to this project will be documented in this file.
 
 - 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.
   - Labels and selectors for agents.
 
 [#1290]: https://github.com/stackabletech/operator-rs/pull/1290
diff --git a/crates/stackable-operator/src/commons/platform_access/anonymous.rs b/crates/stackable-operator/src/commons/platform_access/anonymous.rs
new file mode 100644
index 000000000..439819047
--- /dev/null
+++ b/crates/stackable-operator/src/commons/platform_access/anonymous.rs
@@ -0,0 +1,27 @@
+use schemars::JsonSchema;
+use serde::{Deserialize, Serialize};
+
+/// 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::*;
+
+    #[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/commons/platform_access/mod.rs b/crates/stackable-operator/src/commons/platform_access/mod.rs
index 7cedc76bf..421d905c7 100644
--- a/crates/stackable-operator/src/commons/platform_access/mod.rs
+++ b/crates/stackable-operator/src/commons/platform_access/mod.rs
@@ -1,3 +1,4 @@
-//! Credentials a product cluster grants the platform-access agent, and how to mount them.
+//! Credentials a product cluster grants the platform-access agent, or none, and how to mount them.
 
+pub mod anonymous;
 pub mod tls;

From e364acdf68670d9e8980a85a837a1a9b85b80922 Mon Sep 17 00:00:00 2001
From: Benedikt Labrenz <benedikt@labrenz.org>
Date: Tue, 6 Oct 2026 17:54:14 +0200
Subject: [PATCH 7/9] make crd fragments versioned

---
 crates/stackable-operator/src/commons/mod.rs  |  1 -
 .../src/commons/platform_access/mod.rs        |  4 ---
 crates/stackable-operator/src/crd/mod.rs      |  1 +
 .../platform_access/anonymous/mod.rs}         | 13 +++++---
 .../src/crd/platform_access/mod.rs            |  8 +++++
 .../src/crd/platform_access/tls/mod.rs        | 32 +++++++++++++++++++
 .../platform_access/tls/v1alpha1_impl.rs}     | 25 ++-------------
 7 files changed, 52 insertions(+), 32 deletions(-)
 delete mode 100644 crates/stackable-operator/src/commons/platform_access/mod.rs
 rename crates/stackable-operator/src/{commons/platform_access/anonymous.rs => crd/platform_access/anonymous/mod.rs} (63%)
 create mode 100644 crates/stackable-operator/src/crd/platform_access/mod.rs
 create mode 100644 crates/stackable-operator/src/crd/platform_access/tls/mod.rs
 rename crates/stackable-operator/src/{commons/platform_access/tls.rs => crd/platform_access/tls/v1alpha1_impl.rs} (80%)

diff --git a/crates/stackable-operator/src/commons/mod.rs b/crates/stackable-operator/src/commons/mod.rs
index 8da787e4d..cc4ab30e3 100644
--- a/crates/stackable-operator/src/commons/mod.rs
+++ b/crates/stackable-operator/src/commons/mod.rs
@@ -6,7 +6,6 @@ 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/mod.rs b/crates/stackable-operator/src/commons/platform_access/mod.rs
deleted file mode 100644
index 421d905c7..000000000
--- a/crates/stackable-operator/src/commons/platform_access/mod.rs
+++ /dev/null
@@ -1,4 +0,0 @@
-//! Credentials a product cluster grants the platform-access agent, or none, and how to mount them.
-
-pub mod anonymous;
-pub mod tls;
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/commons/platform_access/anonymous.rs b/crates/stackable-operator/src/crd/platform_access/anonymous/mod.rs
similarity index 63%
rename from crates/stackable-operator/src/commons/platform_access/anonymous.rs
rename to crates/stackable-operator/src/crd/platform_access/anonymous/mod.rs
index 439819047..04b52eae6 100644
--- a/crates/stackable-operator/src/commons/platform_access/anonymous.rs
+++ b/crates/stackable-operator/src/crd/platform_access/anonymous/mod.rs
@@ -1,13 +1,18 @@
 use schemars::JsonSchema;
 use serde::{Deserialize, Serialize};
 
-/// No credential: the product does not authenticate the agent.
-#[derive(Clone, Debug, Default, Deserialize, Eq, JsonSchema, PartialEq, Serialize)]
-pub struct Anonymous {}
+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::*;
+    use super::v1alpha1::Anonymous;
 
     #[test]
     fn deserialize_empty_object() {
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..24b92efae
--- /dev/null
+++ b/crates/stackable-operator/src/crd/platform_access/mod.rs
@@ -0,0 +1,8 @@
+//! CRD fragments for resources managed by product agents, and for the credentials the agents use.
+
+pub mod anonymous;
+pub mod tls;
+
+pub mod v1alpha1 {
+    pub use super::{anonymous::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/commons/platform_access/tls.rs b/crates/stackable-operator/src/crd/platform_access/tls/v1alpha1_impl.rs
similarity index 80%
rename from crates/stackable-operator/src/commons/platform_access/tls.rs
rename to crates/stackable-operator/src/crd/platform_access/tls/v1alpha1_impl.rs
index 3f383561b..936cebf16 100644
--- a/crates/stackable-operator/src/commons/platform_access/tls.rs
+++ b/crates/stackable-operator/src/crd/platform_access/tls/v1alpha1_impl.rs
@@ -1,7 +1,4 @@
-use const_format::concatcp;
 use k8s_openapi::api::core::v1::{SecretVolumeSource, Volume, VolumeMount};
-use schemars::JsonSchema;
-use serde::{Deserialize, Serialize};
 
 use crate::{
     builder::pod::{
@@ -12,28 +9,9 @@ use crate::{
         },
     },
     commons::secret_class::SecretClassVolumeProvisionParts,
-    constants::secret::SECRET_BASE_PATH,
-    v2::types::kubernetes::{SecretClassName, SecretName},
+    crd::platform_access::tls::{MOUNT_PATH, VOLUME_NAME, v1alpha1::TlsClientCredential},
 };
 
-/// 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);
-
-/// 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),
-}
-
 impl TlsClientCredential {
     /// Adds the certificate volume to the Pod and mounts it into all given containers.
     /// - TlsClientCredential::Secret mounts the Secret
@@ -85,6 +63,7 @@ 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() {

From 768793881ec1997826f338a6596a780126d0bfab Mon Sep 17 00:00:00 2001
From: Benedikt Labrenz <benedikt@labrenz.org>
Date: Wed, 7 Oct 2026 11:35:29 +0200
Subject: [PATCH 8/9] add ManagementPolicy CRD fragment

---
 crates/stackable-operator/CHANGELOG.md        |  2 +
 crates/stackable-operator/src/commons/mod.rs  |  1 +
 .../src/commons/platform_access/mod.rs        |  4 ++
 .../src/commons/platform_access/ownership.rs  | 37 ++++++++++++++++
 .../src/crd/platform_access/management/mod.rs | 44 +++++++++++++++++++
 .../management/v1alpha1_impl.rs               | 44 +++++++++++++++++++
 .../src/crd/platform_access/mod.rs            |  3 +-
 7 files changed, 134 insertions(+), 1 deletion(-)
 create mode 100644 crates/stackable-operator/src/commons/platform_access/mod.rs
 create mode 100644 crates/stackable-operator/src/commons/platform_access/ownership.rs
 create mode 100644 crates/stackable-operator/src/crd/platform_access/management/mod.rs
 create mode 100644 crates/stackable-operator/src/crd/platform_access/management/v1alpha1_impl.rs

diff --git a/crates/stackable-operator/CHANGELOG.md b/crates/stackable-operator/CHANGELOG.md
index 0d862e6fd..8bf6d49e9 100644
--- a/crates/stackable-operator/CHANGELOG.md
+++ b/crates/stackable-operator/CHANGELOG.md
@@ -9,6 +9,8 @@ All notable changes to this project will be documented in this file.
 - 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.
+  - Annotation marking resources owned by an agent.
   - Labels and selectors for agents.
 
 [#1290]: https://github.com/stackabletech/operator-rs/pull/1290
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/mod.rs b/crates/stackable-operator/src/commons/platform_access/mod.rs
new file mode 100644
index 000000000..d369b29a0
--- /dev/null
+++ b/crates/stackable-operator/src/commons/platform_access/mod.rs
@@ -0,0 +1,4 @@
+//! Helpers for agents managing resources inside a product. The CRD fragments live in
+//! [`crate::crd::platform_access`].
+
+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/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
index 24b92efae..e116b7317 100644
--- a/crates/stackable-operator/src/crd/platform_access/mod.rs
+++ b/crates/stackable-operator/src/crd/platform_access/mod.rs
@@ -1,8 +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::*, tls::v1alpha1::*};
+    pub use super::{anonymous::v1alpha1::*, management::v1alpha1::*, tls::v1alpha1::*};
 }

From 754e444b8cc3daf6a59fb69d7eaf632249ccc3be Mon Sep 17 00:00:00 2001
From: Benedikt Labrenz <benedikt@labrenz.org>
Date: Wed, 7 Oct 2026 11:40:30 +0200
Subject: [PATCH 9/9] add Ready & Synced conditions for resources reconciled by
 agents

---
 crates/stackable-operator/CHANGELOG.md        |   1 +
 .../src/commons/platform_access/condition.rs  | 144 ++++++++++++++++++
 .../src/commons/platform_access/mod.rs        |   1 +
 3 files changed, 146 insertions(+)
 create mode 100644 crates/stackable-operator/src/commons/platform_access/condition.rs

diff --git a/crates/stackable-operator/CHANGELOG.md b/crates/stackable-operator/CHANGELOG.md
index 8bf6d49e9..c1249063c 100644
--- a/crates/stackable-operator/CHANGELOG.md
+++ b/crates/stackable-operator/CHANGELOG.md
@@ -10,6 +10,7 @@ All notable changes to this project will be documented in this file.
   - `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.
 
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
index d369b29a0..d557ad887 100644
--- a/crates/stackable-operator/src/commons/platform_access/mod.rs
+++ b/crates/stackable-operator/src/commons/platform_access/mod.rs
@@ -1,4 +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;