diff --git a/ci/nightly/pipeline.template.yml b/ci/nightly/pipeline.template.yml
index 3c418f9e46c13..d7ff95fd6b695 100644
--- a/ci/nightly/pipeline.template.yml
+++ b/ci/nightly/pipeline.template.yml
@@ -2763,6 +2763,20 @@ steps:
         agents:
           queue: hetzner-aarch64-16cpu-32gb
 
+      - id: orchestratord-failure-events
+        topics: [self-managed]
+        label: "Orchestratord reconciliation failure event tests"
+        artifact_paths: ["mz_debug_*.zip"]
+        depends_on: devel-docker-tags
+        timeout_in_minutes: 60
+        plugins:
+          - ./ci/plugins/mzcompose:
+              composition: orchestratord
+              run: failure-events
+              ci-builder: stable
+        agents:
+          queue: hetzner-aarch64-16cpu-32gb
+
       - id: orchestratord-manually-promote
         topics: [0dt-migration, self-managed]
         label: "Orchestratord ManuallyPromote tests"
diff --git a/doc/user/data/metrics.yml b/doc/user/data/metrics.yml
index d42be02e4edd0..a05631db6105e 100644
--- a/doc/user/data/metrics.yml
+++ b/doc/user/data/metrics.yml
@@ -4381,6 +4381,66 @@ metrics:
   help: Whether this operator replica holds the controller leadership lease, and is therefore the replica reconciling. Summed across the replicas this should be 1. A sustained 0 means no replica can take the lease, for instance because the service account lacks permission on leases, and the operator is reconciling nothing.
   source: src/orchestratord/src/metrics.rs
   visibility: internal
+- name: orchestratord_reconciliation_duration_seconds_bucket
+  help: Time spent in one reconciliation pass. This covers only the reconciler itself, not the finalizer bookkeeping around it, and a pass that waits on a rollout returns promptly rather than blocking, so this measures work done and not the wall-clock length of a rollout.
+  labels:
+  - controller
+  - event_type
+  - le
+  source: src/orchestratord/src/reconcile.rs
+  visibility: internal
+- name: orchestratord_reconciliation_duration_seconds_count
+  help: Time spent in one reconciliation pass. This covers only the reconciler itself, not the finalizer bookkeeping around it, and a pass that waits on a rollout returns promptly rather than blocking, so this measures work done and not the wall-clock length of a rollout.
+  labels:
+  - controller
+  - event_type
+  source: src/orchestratord/src/reconcile.rs
+  visibility: internal
+- name: orchestratord_reconciliation_duration_seconds_sum
+  help: Time spent in one reconciliation pass. This covers only the reconciler itself, not the finalizer bookkeeping around it, and a pass that waits on a rollout returns promptly rather than blocking, so this measures work done and not the wall-clock length of a rollout.
+  labels:
+  - controller
+  - event_type
+  source: src/orchestratord/src/reconcile.rs
+  visibility: internal
+- name: orchestratord_reconciliation_step_duration_seconds_bucket
+  help: Time spent in one reconciliation step.
+  labels:
+  - controller
+  - le
+  - step
+  source: src/orchestratord/src/reconcile.rs
+  visibility: internal
+- name: orchestratord_reconciliation_step_duration_seconds_count
+  help: Time spent in one reconciliation step.
+  labels:
+  - controller
+  - step
+  source: src/orchestratord/src/reconcile.rs
+  visibility: internal
+- name: orchestratord_reconciliation_step_duration_seconds_sum
+  help: Time spent in one reconciliation step.
+  labels:
+  - controller
+  - step
+  source: src/orchestratord/src/reconcile.rs
+  visibility: internal
+- name: orchestratord_reconciliation_steps_total
+  help: Count of reconciliation steps, by controller, by step, and by what the step concluded. Steps are the named phases a reconciliation pass moves through, so this is where a pass that fails or stalls identifies which phase it was in. An `outcome` of `abandoned` means the step did not conclude, which covers both an error propagating out of it and the pass being cancelled by a leadership handoff, so alert on orchestratord_reconciliations_total instead and read this to locate the step.
+  labels:
+  - controller
+  - outcome
+  - step
+  source: src/orchestratord/src/reconcile.rs
+  visibility: internal
+- name: orchestratord_reconciliations_total
+  help: Count of reconciliation passes, by controller, by which of the controller's entry points ran, and by what the pass concluded. An `outcome` of `waiting` means the pass asked to be retried, which is normal during a rollout; `failed` means it returned an error, and is also reported as a Kubernetes event on the resource.
+  labels:
+  - controller
+  - event_type
+  - outcome
+  source: src/orchestratord/src/reconcile.rs
+  visibility: internal
 - name: outer_join_lowering_cases
   help: How many times the different outer join lowering cases happened.
   labels:
diff --git a/misc/helm-charts/operator/templates/clusterrole.yaml b/misc/helm-charts/operator/templates/clusterrole.yaml
index 23cc4c105112a..f428c81be6089 100644
--- a/misc/helm-charts/operator/templates/clusterrole.yaml
+++ b/misc/helm-charts/operator/templates/clusterrole.yaml
@@ -37,6 +37,16 @@ rules:
   - get
   - list
   - watch
+# The controllers publish a warning event on the Materialize, Balancer, or
+# Console resource whose reconciliation failed, so that `kubectl describe`
+# explains a stuck object without needing the operator's logs. Patch is for
+# the aggregation of a repeated failure into one event with a count.
+- apiGroups: ["events.k8s.io"]
+  resources:
+  - events
+  verbs:
+  - create
+  - patch
 - apiGroups: ["coordination.k8s.io"]
   resources:
   - leases
diff --git a/misc/helm-charts/operator/tests/clusterrole_test.yaml b/misc/helm-charts/operator/tests/clusterrole_test.yaml
index 1e23f034d2558..cf6d940a3939c 100644
--- a/misc/helm-charts/operator/tests/clusterrole_test.yaml
+++ b/misc/helm-charts/operator/tests/clusterrole_test.yaml
@@ -32,6 +32,15 @@ tests:
             apiGroups: ["rbac.authorization.k8s.io"]
             resources: ["roles", "rolebindings"]
             verbs: ["create", "update", "patch", "delete", "get", "list", "watch"]
+      # Without this the operator reconciles fine but cannot report a failed
+      # reconciliation on the resource it failed on, which is the only place a
+      # user without log access sees it.
+      - contains:
+          path: rules
+          content:
+            apiGroups: ["events.k8s.io"]
+            resources: ["events"]
+            verbs: ["create", "patch"]
 
   - it: should not create a clusterrole when RBAC is disabled
     set:
diff --git a/src/orchestratord/src/bin/orchestratord.rs b/src/orchestratord/src/bin/orchestratord.rs
index 59caf93d713c9..c891a72ecd1a5 100644
--- a/src/orchestratord/src/bin/orchestratord.rs
+++ b/src/orchestratord/src/bin/orchestratord.rs
@@ -45,6 +45,7 @@ use mz_orchestratord::{
     controller, gcp_node_upgrade,
     k8s::{ConversionWebhookConfig, register_crds},
     metrics::{self, Metrics},
+    reconcile::{Observed, Publisher},
     tls::DefaultCertificateSpecs,
     webhook,
 };
@@ -631,6 +632,13 @@ async fn run(args: Args) -> Result<(), anyhow::Error> {
         &leader_election_identity,
     );
 
+    // Shared by all of the controllers, so that the aggregation of repeated
+    // events is too.
+    let publisher = Arc::new(Publisher::new(
+        client.clone(),
+        leader_election_identity.clone(),
+    ));
+
     // Each of these is rebuilt every time this replica wins the election, since
     // running a controller consumes it and we rejoin the election after losing
     // the lease.
@@ -708,10 +716,16 @@ async fn run(args: Args) -> Result<(), anyhow::Error> {
             tracing: args.tracing,
             orchestratord_namespace: namespace,
         };
+        let publisher = Arc::clone(&publisher);
         move || {
             k8s_controller::Controller::namespaced_all(
                 client.clone(),
-                controller::materialize::Context::new(config.clone(), Arc::clone(&metrics)),
+                Observed::new(
+                    controller::materialize::Context::new(config.clone(), Arc::clone(&metrics)),
+                    controller::materialize::CONTROLLER_NAME,
+                    Arc::clone(&metrics.reconcile),
+                    Arc::clone(&publisher),
+                ),
                 watcher::Config::default().timeout(29),
             )
             .with_controller(|controller| {
@@ -771,10 +785,20 @@ async fn run(args: Args) -> Result<(), anyhow::Error> {
             balancerd_http_port: args.balancerd_http_port,
             balancerd_internal_http_port: args.balancerd_internal_http_port,
         };
+        let reconcile_metrics = Arc::clone(&metrics.reconcile);
+        let publisher = Arc::clone(&publisher);
         move || {
             k8s_controller::Controller::namespaced_all(
                 client.clone(),
-                controller::balancer::Context::new(config.clone()),
+                Observed::new(
+                    controller::balancer::Context::new(
+                        config.clone(),
+                        Arc::clone(&reconcile_metrics),
+                    ),
+                    controller::balancer::CONTROLLER_NAME,
+                    Arc::clone(&reconcile_metrics),
+                    Arc::clone(&publisher),
+                ),
                 watcher::Config::default().timeout(29),
             )
             .with_controller(|controller| {
@@ -821,10 +845,20 @@ async fn run(args: Args) -> Result<(), anyhow::Error> {
             console_http_port: args.console_http_port,
             balancerd_http_port: args.balancerd_http_port,
         };
+        let reconcile_metrics = Arc::clone(&metrics.reconcile);
+        let publisher = Arc::clone(&publisher);
         move || {
             k8s_controller::Controller::namespaced_all(
                 client.clone(),
-                controller::console::Context::new(config.clone()),
+                Observed::new(
+                    controller::console::Context::new(
+                        config.clone(),
+                        Arc::clone(&reconcile_metrics),
+                    ),
+                    controller::console::CONTROLLER_NAME,
+                    Arc::clone(&reconcile_metrics),
+                    Arc::clone(&publisher),
+                ),
                 watcher::Config::default().timeout(29),
             )
             .with_controller(|controller| {
diff --git a/src/orchestratord/src/controller/balancer.rs b/src/orchestratord/src/controller/balancer.rs
index 367151240b5d0..d844c34cf6100 100644
--- a/src/orchestratord/src/controller/balancer.rs
+++ b/src/orchestratord/src/controller/balancer.rs
@@ -7,6 +7,8 @@
 // the Business Source License, use of this software will be governed
 // by the Apache License, Version 2.0.
 
+use std::sync::Arc;
+
 use anyhow::bail;
 use k8s_controller::TraceMetadata;
 use k8s_openapi::{
@@ -36,6 +38,7 @@ use tracing::{trace, warn};
 use crate::{
     Error,
     k8s::{apply_resource, get_resource, recommended_k8s_labels, replace_resource},
+    reconcile::{self, Outcome},
     tls::{DefaultCertificateSpecs, create_certificate, issuer_ref_defined},
 };
 use mz_cloud_resources::crd::{
@@ -46,6 +49,11 @@ use mz_cloud_resources::crd::{
 use mz_orchestrator_kubernetes::KubernetesImagePullPolicy;
 use mz_ore::{cli::KeyValueArg, instrument};
 
+/// The `controller` label that this controller's metrics and Kubernetes events
+/// carry. Shared by `Observed` and by every step this controller records, which
+/// must agree for a pass and its steps to line up.
+pub const CONTROLLER_NAME: &str = "balancer";
+
 #[derive(Clone)]
 pub struct Config {
     pub enable_security_context: bool,
@@ -69,11 +77,18 @@ pub struct Config {
 
 pub struct Context {
     config: Config,
+    metrics: Arc<reconcile::Metrics>,
 }
 
 impl Context {
-    pub fn new(config: Config) -> Self {
-        Self { config }
+    pub fn new(config: Config, metrics: Arc<reconcile::Metrics>) -> Self {
+        Self { config, metrics }
+    }
+
+    /// Starts recording the step named `step` of this controller's
+    /// reconciliation.
+    fn step(&self, step: &'static str) -> reconcile::Step<'_> {
+        self.metrics.step(CONTROLLER_NAME, step)
     }
 
     async fn sync_deployment_status(
@@ -562,6 +577,7 @@ impl k8s_controller::Context for Context {
         _metadata: &mut TraceMetadata,
     ) -> Result<Option<Action>, Self::Error> {
         if balancer.status.is_none() {
+            let step = self.step("initialize_status");
             let balancer_api: Api<Balancer> =
                 Api::namespaced(client.clone(), &balancer.meta().namespace.clone().unwrap());
             let mut new_balancer = balancer.clone();
@@ -573,6 +589,7 @@ impl k8s_controller::Context for Context {
                     &new_balancer,
                 )
                 .await?;
+            step.finish(Outcome::Applied);
             // Updating the status should trigger a reconciliation
             // which will include a status this time.
             return Ok(None);
@@ -583,21 +600,31 @@ impl k8s_controller::Context for Context {
         let deployment_api: Api<Deployment> = Api::namespaced(client.clone(), &namespace);
         let service_api: Api<Service> = Api::namespaced(client.clone(), &namespace);
 
+        let step = self.step("certificate");
         if let Some(external_certificate) = self.create_external_certificate_object(balancer)? {
             trace!("creating new balancerd external certificate");
             apply_resource(&certificate_api, &external_certificate).await?;
+            step.finish(Outcome::Applied);
+        } else {
+            step.finish(Outcome::Skipped);
         }
 
+        let step = self.step("deployment");
         let deployment = self.create_deployment_object(balancer)?;
         self.fix_deployment(&deployment_api, &deployment).await?;
         trace!("creating new balancerd deployment");
         apply_resource(&deployment_api, &deployment).await?;
+        step.finish(Outcome::Applied);
 
+        let step = self.step("service");
         let service = self.create_service_object(balancer);
         trace!("creating new balancerd service");
         apply_resource(&service_api, &service).await?;
+        step.finish(Outcome::Applied);
 
+        let step = self.step("sync_status");
         self.sync_deployment_status(&client, balancer).await?;
+        step.finish(Outcome::Applied);
 
         Ok(None)
     }
diff --git a/src/orchestratord/src/controller/console.rs b/src/orchestratord/src/controller/console.rs
index 173bef895c2e1..0c330f4075606 100644
--- a/src/orchestratord/src/controller/console.rs
+++ b/src/orchestratord/src/controller/console.rs
@@ -7,6 +7,8 @@
 // the Business Source License, use of this software will be governed
 // by the Apache License, Version 2.0.
 
+use std::sync::Arc;
+
 use k8s_controller::TraceMetadata;
 use k8s_openapi::{
     api::{
@@ -40,6 +42,7 @@ use tracing::{trace, warn};
 use crate::{
     Error,
     k8s::{apply_resource, get_resource, recommended_k8s_labels, replace_resource},
+    reconcile::{self, Outcome},
     tls::{DefaultCertificateSpecs, create_certificate, issuer_ref_defined},
 };
 use mz_cloud_resources::crd::{
@@ -51,6 +54,11 @@ use mz_orchestrator_kubernetes::KubernetesImagePullPolicy;
 use mz_ore::{cli::KeyValueArg, instrument};
 use mz_server_core::listeners::AuthenticatorKind;
 
+/// The `controller` label that this controller's metrics and Kubernetes events
+/// carry. Shared by `Observed` and by every step this controller records, which
+/// must agree for a pass and its steps to line up.
+pub const CONTROLLER_NAME: &str = "console";
+
 #[derive(Clone)]
 pub struct Config {
     pub enable_security_context: bool,
@@ -86,11 +94,18 @@ struct AppConfigAuth {
 
 pub struct Context {
     config: Config,
+    metrics: Arc<reconcile::Metrics>,
 }
 
 impl Context {
-    pub fn new(config: Config) -> Self {
-        Self { config }
+    pub fn new(config: Config, metrics: Arc<reconcile::Metrics>) -> Self {
+        Self { config, metrics }
+    }
+
+    /// Starts recording the step named `step` of this controller's
+    /// reconciliation.
+    fn step(&self, step: &'static str) -> reconcile::Step<'_> {
+        self.metrics.step(CONTROLLER_NAME, step)
     }
 
     async fn sync_deployment_status(
@@ -590,6 +605,7 @@ impl k8s_controller::Context for Context {
         _metadata: &mut TraceMetadata,
     ) -> Result<Option<Action>, Self::Error> {
         if console.status.is_none() {
+            let step = self.step("initialize_status");
             let console_api: Api<Console> =
                 Api::namespaced(client.clone(), &console.meta().namespace.clone().unwrap());
             let mut new_console = console.clone();
@@ -601,6 +617,7 @@ impl k8s_controller::Context for Context {
                     &new_console,
                 )
                 .await?;
+            step.finish(Outcome::Applied);
             // Updating the status should trigger a reconciliation
             // which will include a status this time.
             return Ok(None);
@@ -613,33 +630,51 @@ impl k8s_controller::Context for Context {
         let service_api: Api<Service> = Api::namespaced(client.clone(), &namespace);
         let certificate_api: Api<Certificate> = Api::namespaced(client.clone(), &namespace);
 
+        let step = self.step("network_policies");
         trace!("creating new network policies");
         let network_policies = self.create_network_policies(console);
         for network_policy in &network_policies {
             apply_resource(&network_policy_api, network_policy).await?;
         }
+        step.finish(if network_policies.is_empty() {
+            Outcome::Skipped
+        } else {
+            Outcome::Applied
+        });
 
+        let step = self.step("configmap");
         trace!("creating new console configmap");
         let console_configmap = self.create_console_app_configmap_object(console);
         apply_resource(&configmap_api, &console_configmap).await?;
+        step.finish(Outcome::Applied);
 
+        let step = self.step("deployment");
         trace!("creating new console deployment");
         let console_deployment = self.create_console_deployment_object(console);
         self.fix_deployment(&deployment_api, &console_deployment)
             .await?;
         apply_resource(&deployment_api, &console_deployment).await?;
+        step.finish(Outcome::Applied);
 
+        let step = self.step("service");
         trace!("creating new console service");
         let console_service = self.create_console_service_object(console);
         apply_resource(&service_api, &console_service).await?;
+        step.finish(Outcome::Applied);
 
+        let step = self.step("certificate");
         let console_external_certificate = self.create_console_external_certificate(console)?;
         if let Some(certificate) = &console_external_certificate {
             trace!("creating new console external certificate");
             apply_resource(&certificate_api, certificate).await?;
+            step.finish(Outcome::Applied);
+        } else {
+            step.finish(Outcome::Skipped);
         }
 
+        let step = self.step("sync_status");
         self.sync_deployment_status(&client, console).await?;
+        step.finish(Outcome::Applied);
 
         Ok(None)
     }
diff --git a/src/orchestratord/src/controller/materialize.rs b/src/orchestratord/src/controller/materialize.rs
index 177b339834b9f..bf44cde278112 100644
--- a/src/orchestratord/src/controller/materialize.rs
+++ b/src/orchestratord/src/controller/materialize.rs
@@ -36,6 +36,7 @@ use crate::{
     matching_image_from_environmentd_image_ref,
     metrics::Metrics,
     parse_image_tag,
+    reconcile::{self, Outcome},
     tls::{DefaultCertificateSpecs, issuer_ref_defined, resolved_dns_names},
 };
 use mz_cloud_provider::CloudProvider;
@@ -54,6 +55,11 @@ use mz_ore::{cast::CastFrom, cli::KeyValueArg, instrument};
 pub mod generation;
 pub mod global;
 
+/// The `controller` label that this controller's metrics and Kubernetes events
+/// carry. Shared by `Observed` and by every step this controller records, which
+/// must agree for a pass and its steps to line up.
+pub const CONTROLLER_NAME: &str = "materialize";
+
 #[derive(Clone)]
 pub struct Config {
     pub cloud_provider: CloudProvider,
@@ -150,6 +156,29 @@ impl Context {
         }
     }
 
+    /// Starts recording the step named `step` of this controller's
+    /// reconciliation.
+    fn step(&self, step: &'static str) -> reconcile::Step<'_> {
+        self.metrics.reconcile.step(CONTROLLER_NAME, step)
+    }
+
+    /// Deletes `generation`'s resources, releasing the read holds its
+    /// environmentd was keeping.
+    async fn teardown_generation(
+        &self,
+        client: &Client,
+        mz: &Materialize,
+        resources: &generation::Resources,
+        generation: u64,
+    ) -> Result<(), Error> {
+        let step = self.step("teardown_generation");
+        resources
+            .teardown_generation(client, mz, generation)
+            .await?;
+        step.finish(Outcome::Applied);
+        Ok(())
+    }
+
     fn set_needs_update(&self, mz: &Materialize, needs_update: bool) {
         let mut needs_update_set = self.needs_update.lock().unwrap();
         if needs_update {
@@ -195,11 +224,14 @@ impl Context {
         desired_generation: u64,
         resources_hash: String,
     ) -> Result<Option<Action>, Error> {
+        let step = self.step("promote");
         if let Some(action) = resources.promote_services(client, &mz.namespace()).await? {
+            step.finish(Outcome::Waiting);
             return Ok(Some(action));
         }
-        resources
-            .teardown_generation(client, mz, active_generation)
+        step.finish(Outcome::Applied);
+
+        self.teardown_generation(client, mz, &resources, active_generation)
             .await?;
         let mz_api: Api<Materialize> = Api::namespaced(client.clone(), &mz.namespace());
         self.update_status(
@@ -287,12 +319,19 @@ impl k8s_controller::Context for Context {
 
         let status = mz.status();
         if mz.status.is_none() {
+            let step = self.step("initialize_status");
             self.update_status(&mz_api, mz, status, true).await?;
+            step.finish(Outcome::Applied);
             // Updating the status should trigger a reconciliation
             // which will include a status this time.
             return Ok(None);
         }
 
+        // Everything from reading the license key through the uniqueness check
+        // is one step: it is all about settling on the environment id, and it
+        // is where a misconfigured license key stops a new environment before
+        // any of its resources exist.
+        let step = self.step("resolve_environment_id");
         let backend_secret = secret_api.get(&mz.spec.backend_secret_name).await?;
         let license_key_environment_id: Option<Uuid> = if let Some(license_key) = backend_secret
             .data
@@ -348,6 +387,7 @@ impl k8s_controller::Context for Context {
             mz_api
                 .replace(&mz.name_unchecked(), &PostParams::default(), &mz)
                 .await?;
+            step.finish(Outcome::Applied);
             // Updating the spec should also trigger a reconciliation.
             // We can't do that as part of the above check because you can't
             // update both the spec and the status in a single api call.
@@ -368,10 +408,13 @@ impl k8s_controller::Context for Context {
         }
 
         self.check_environment_id_conflicts(&client, mz).await?;
+        step.finish(Outcome::Applied);
 
+        let step = self.step("global_resources");
         global::Resources::new(&self.config, mz)?
             .apply(&client, &mz.namespace())
             .await?;
+        step.finish(Outcome::Applied);
 
         // we compare the hash against the environment resources generated
         // for the current active generation, since that's what we expect to
@@ -437,8 +480,7 @@ impl k8s_controller::Context for Context {
                             );
                             // Tear down the un-promoted generation to release
                             // its read holds.
-                            resources
-                                .teardown_generation(&client, mz, next_generation)
+                            self.teardown_generation(&client, mz, &resources, next_generation)
                                 .await?;
                             self.update_status(
                                 &mz_api,
@@ -573,16 +615,17 @@ impl k8s_controller::Context for Context {
                     // The only reason someone would choose this strategy is if they didn't have
                     // space for the two generations of pods.
                     // Lets make room for the new ones by deleting the old generation.
-                    resources
-                        .teardown_generation(&client, mz, active_generation)
+                    self.teardown_generation(&client, mz, &resources, active_generation)
                         .await?;
                 }
 
                 trace!("applying environment resources");
-                match resources
+                let step = self.step("generation_resources");
+                let applied = resources
                     .apply(&client, mz.should_force_promote(), &mz.namespace())
-                    .await
-                {
+                    .await;
+                step.finish_with(&applied);
+                match applied {
                     Ok(Some(action)) => {
                         trace!("new environment is not yet ready");
                         Ok(Some(action))
@@ -717,8 +760,7 @@ impl k8s_controller::Context for Context {
             (false, true, false) => {
                 let mut needs_update = mz.conditions_need_update();
                 if mz.update_in_progress() {
-                    resources
-                        .teardown_generation(&client, mz, next_generation)
+                    self.teardown_generation(&client, mz, &resources, next_generation)
                         .await?;
                     needs_update = true;
                 }
@@ -760,8 +802,7 @@ impl k8s_controller::Context for Context {
                 // WaitingForApproval.
                 let mut needs_update = mz.conditions_need_update() || mz.rollout_requested();
                 if mz.update_in_progress() {
-                    resources
-                        .teardown_generation(&client, mz, next_generation)
+                    self.teardown_generation(&client, mz, &resources, next_generation)
                         .await?;
                     needs_update = true;
                 }
@@ -805,6 +846,7 @@ impl k8s_controller::Context for Context {
         // enforced by the environmentd rollout process being able to call
         // into the promotion endpoint
 
+        let step = self.step("balancer");
         if self.config.create_balancers {
             let balancer = Balancer {
                 metadata: mz.managed_resource_meta(mz.name_unchecked()),
@@ -833,8 +875,14 @@ impl k8s_controller::Context for Context {
             };
             let balancer = apply_resource(&balancer_api, &balancer).await?;
             result = wait_for_balancer(&balancer)?;
+            step.finish(match result {
+                // The balancer is not ready yet, so we will be back.
+                Some(_) => Outcome::Waiting,
+                None => Outcome::Applied,
+            });
         } else {
             delete_resource(&balancer_api, &mz.name_unchecked()).await?;
+            step.finish(Outcome::Skipped);
         }
 
         if let Some(action) = result {
@@ -844,6 +892,7 @@ impl k8s_controller::Context for Context {
         // and the console relies on the balancer service existing, which is
         // enforced by wait_for_balancer
 
+        let step = self.step("console");
         if self.config.create_console {
             let active_environmentd_image_ref = mz.active_environmentd_image_ref();
             let environmentd_image_tag =
@@ -890,8 +939,10 @@ impl k8s_controller::Context for Context {
                 status: None,
             };
             apply_resource(&console_api, &console).await?;
+            step.finish(Outcome::Applied);
         } else {
             delete_resource(&console_api, &mz.name_unchecked()).await?;
+            step.finish(Outcome::Skipped);
         }
 
         Ok(result)
diff --git a/src/orchestratord/src/lib.rs b/src/orchestratord/src/lib.rs
index 645e5928c302a..ff724dceef4f4 100644
--- a/src/orchestratord/src/lib.rs
+++ b/src/orchestratord/src/lib.rs
@@ -13,6 +13,7 @@ pub mod controller;
 pub mod gcp_node_upgrade;
 pub mod k8s;
 pub mod metrics;
+pub mod reconcile;
 pub mod tls;
 pub mod webhook;
 
diff --git a/src/orchestratord/src/metrics.rs b/src/orchestratord/src/metrics.rs
index 3125d7f7f41a1..17b6d21e7b312 100644
--- a/src/orchestratord/src/metrics.rs
+++ b/src/orchestratord/src/metrics.rs
@@ -8,6 +8,7 @@
 // by the Apache License, Version 2.0.
 
 use std::collections::BTreeMap;
+use std::sync::Arc;
 use std::time::Duration;
 
 use axum::{Extension, Router, body::Body, routing::get};
@@ -19,10 +20,16 @@ use tracing::{Level, Span};
 use mz_ore::metric;
 use mz_ore::metrics::{MetricsRegistry, UIntGauge};
 
+use crate::reconcile;
+
 #[derive(Debug)]
 pub struct Metrics {
     pub is_leader: UIntGauge,
     pub environmentd_needs_update: UIntGauge,
+    /// Metrics covering the reconciliation loop itself, shared by every
+    /// controller. Held behind an `Arc` because the controllers and the
+    /// wrappers that observe them each need a handle.
+    pub reconcile: Arc<reconcile::Metrics>,
 }
 
 impl Metrics {
@@ -38,6 +45,7 @@ impl Metrics {
                     name: "environmentd_needs_update",
                     help: "Count of organizations in this cluster which are running outdated pod templates. Only the operator replica holding the leadership lease reconciles, so the others report zero.",
                 }),
+            reconcile: Arc::new(reconcile::Metrics::register_into(registry)),
         }
     }
 
diff --git a/src/orchestratord/src/reconcile.rs b/src/orchestratord/src/reconcile.rs
new file mode 100644
index 0000000000000..2439193c06b91
--- /dev/null
+++ b/src/orchestratord/src/reconcile.rs
@@ -0,0 +1,561 @@
+// Copyright Materialize, Inc. and contributors. All rights reserved.
+//
+// Use of this software is governed by the Business Source License
+// included in the LICENSE file.
+//
+// As of the Change Date specified in that file, in accordance with
+// the Business Source License, use of this software will be governed
+// by the Apache License, Version 2.0.
+
+//! Observability for the reconciliation loop, shared by every controller in
+//! this process.
+//!
+//! [`Observed`] wraps a controller's [`k8s_controller::Context`] so that each
+//! reconciliation pass is timed and counted, and so that a failed pass is
+//! published as a Kubernetes warning event on the resource being reconciled.
+//! The event is what makes a stuck object explain itself: `kubectl describe`
+//! shows why reconciliation is not progressing to someone who cannot read the
+//! operator's logs.
+//!
+//! Reconcilers additionally report their own [steps](Metrics::step). A step is
+//! the only level at which "there was nothing to do" is distinguishable from
+//! "the desired state was reached", since the reconciler interface reports both
+//! as success.
+
+use std::collections::BTreeMap;
+use std::sync::{Arc, Mutex};
+use std::time::Instant;
+
+use k8s_controller::TraceMetadata;
+use kube::runtime::controller::Action;
+use kube::runtime::events::{Event, EventType, Recorder, Reporter};
+use kube::{Client, Resource};
+use tracing::warn;
+
+use mz_ore::error::ErrorExt;
+use mz_ore::metric;
+use mz_ore::metrics::{
+    MetricsRegistry,
+    raw::{HistogramVec, IntCounterVec},
+};
+
+/// The `reportingController` that events published by this process carry.
+const EVENT_REPORTER: &str = "orchestratord.materialize.cloud";
+
+/// The `reason` of the event published when a reconciliation fails.
+const RECONCILIATION_FAILED: &str = "ReconciliationFailed";
+
+/// The Kubernetes API server rejects an event whose note exceeds 1kB, so a
+/// long error is truncated to fit rather than costing us the event entirely.
+const MAX_NOTE_BYTES: usize = 1024;
+
+/// Buckets shared by both duration histograms, so that a step's latency reads
+/// against the whole pass it belongs to.
+///
+/// Reconciliation is a handful of Kubernetes API calls: measured on a settled
+/// operator, steps run 6ms to 90ms, the two that touch a generation's resources
+/// run 230ms to 420ms, and a whole pass runs 58ms to 350ms. Order-of-magnitude
+/// resolution is all that band needs, so these are deliberately coarser than
+/// `mz_ore::stats::histogram_seconds_buckets`, whose power-of-two spacing costs
+/// three times the series for detail no operator question asks for. Anything
+/// past the top bucket is pathological rather than slow, and `+Inf`, `_sum` and
+/// `_count` still account for it.
+const DURATION_BUCKETS: [f64; 6] = [0.01, 0.05, 0.25, 1.0, 5.0, 30.0];
+
+/// What a reconciliation pass, or one step of one, did.
+///
+/// The variants are the label values of the `outcome` label, so a call site
+/// picking between them decides what a dashboard reports.
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+pub enum Outcome {
+    /// Brought the resources it manages to the desired state, or found them
+    /// already there.
+    Applied,
+    /// Made progress, but the desired state is not reached yet, so
+    /// reconciliation must run again. A rollout spends most of its passes here
+    /// while it waits for the new generation's pods to become ready.
+    Waiting,
+    /// Had nothing to act on: what it manages is disabled by configuration, or
+    /// absent from the resource's spec.
+    Skipped,
+    /// Returned an error, observed as one.
+    Failed,
+    /// Did not reach a conclusion. Either an error propagated out of it, or the
+    /// reconciliation running it was cancelled: this process aborts in-flight
+    /// reconciliations when it loses the leadership lease, and again on
+    /// shutdown.
+    ///
+    /// A [`Step`] dropped without being finished records this, because a `Drop`
+    /// cannot tell those two apart. It is therefore not a failure signal:
+    /// `orchestratord_reconciliations_total` is, since it is recorded from the
+    /// reconciler's actual `Result` and a cancelled pass never reaches it. What
+    /// this answers is which step a pass was in when it stopped.
+    Abandoned,
+}
+
+impl Outcome {
+    fn as_str(self) -> &'static str {
+        match self {
+            Outcome::Applied => "applied",
+            Outcome::Waiting => "waiting",
+            Outcome::Skipped => "skipped",
+            Outcome::Failed => "failed",
+            Outcome::Abandoned => "abandoned",
+        }
+    }
+
+    /// Classifies what a reconciler returned. `Ok(None)` means the reconciler
+    /// is done with this resource until something changes, and `Ok(Some(_))`
+    /// means it asked to be run again.
+    fn of_result<E>(result: &Result<Option<Action>, E>) -> Self {
+        match result {
+            Ok(None) => Outcome::Applied,
+            Ok(Some(_)) => Outcome::Waiting,
+            Err(_) => Outcome::Failed,
+        }
+    }
+}
+
+/// Which of a [`k8s_controller::Context`]'s two entry points ran, as the
+/// `event_type` label.
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+enum EntryPoint {
+    Apply,
+    Cleanup,
+}
+
+impl EntryPoint {
+    fn as_str(self) -> &'static str {
+        match self {
+            EntryPoint::Apply => "apply",
+            EntryPoint::Cleanup => "cleanup",
+        }
+    }
+
+    /// The event `action`, which names what was being attempted rather than
+    /// its outcome.
+    fn action(self) -> &'static str {
+        match self {
+            EntryPoint::Apply => "Reconcile",
+            EntryPoint::Cleanup => "Cleanup",
+        }
+    }
+}
+
+/// Metrics covering the reconciliation loop of every controller.
+#[derive(Debug)]
+pub struct Metrics {
+    reconciliations: IntCounterVec,
+    reconciliation_duration_seconds: HistogramVec,
+    steps: IntCounterVec,
+    step_duration_seconds: HistogramVec,
+}
+
+impl Metrics {
+    pub fn register_into(registry: &MetricsRegistry) -> Self {
+        Self {
+            reconciliations: registry.register(metric! {
+                name: "orchestratord_reconciliations_total",
+                help: "Count of reconciliation passes, by controller, by which of the controller's entry points ran, and by what the pass concluded. An `outcome` of `waiting` means the pass asked to be retried, which is normal during a rollout; `failed` means it returned an error, and is also reported as a Kubernetes event on the resource.",
+                var_labels: ["controller", "event_type", "outcome"],
+            }),
+            reconciliation_duration_seconds: registry.register(metric! {
+                name: "orchestratord_reconciliation_duration_seconds",
+                help: "Time spent in one reconciliation pass. This covers only the reconciler itself, not the finalizer bookkeeping around it, and a pass that waits on a rollout returns promptly rather than blocking, so this measures work done and not the wall-clock length of a rollout.",
+                var_labels: ["controller", "event_type"],
+                buckets: DURATION_BUCKETS.to_vec(),
+            }),
+            steps: registry.register(metric! {
+                name: "orchestratord_reconciliation_steps_total",
+                help: "Count of reconciliation steps, by controller, by step, and by what the step concluded. Steps are the named phases a reconciliation pass moves through, so this is where a pass that fails or stalls identifies which phase it was in. An `outcome` of `abandoned` means the step did not conclude, which covers both an error propagating out of it and the pass being cancelled by a leadership handoff, so alert on orchestratord_reconciliations_total instead and read this to locate the step.",
+                var_labels: ["controller", "step", "outcome"],
+            }),
+            step_duration_seconds: registry.register(metric! {
+                name: "orchestratord_reconciliation_step_duration_seconds",
+                help: "Time spent in one reconciliation step.",
+                var_labels: ["controller", "step"],
+                buckets: DURATION_BUCKETS.to_vec(),
+            }),
+        }
+    }
+
+    fn record_reconciliation(
+        &self,
+        controller: &str,
+        entry_point: EntryPoint,
+        outcome: Outcome,
+        duration: f64,
+    ) {
+        self.reconciliations
+            .with_label_values(&[controller, entry_point.as_str(), outcome.as_str()])
+            .inc();
+        self.reconciliation_duration_seconds
+            .with_label_values(&[controller, entry_point.as_str()])
+            .observe(duration);
+    }
+
+    /// Starts timing the step named `step` of `controller`'s reconciliation.
+    ///
+    /// The step is recorded when the returned [`Step`] is finished or dropped.
+    pub fn step<'a>(&'a self, controller: &'static str, step: &'static str) -> Step<'a> {
+        Step {
+            metrics: self,
+            controller,
+            step,
+            outcome: None,
+            start: Instant::now(),
+        }
+    }
+}
+
+/// Times one named step of a reconciliation, recording its duration and
+/// outcome once dropped.
+///
+/// A step dropped without [`Step::finish`] records [`Outcome::Abandoned`], so
+/// the `?` that leaves a step reports where the pass stopped with no
+/// bookkeeping at the point of the early return. The contract that buys is
+/// that every path out of a step's scope which reaches a conclusion must
+/// finish the step explicitly.
+#[must_use = "a step that is never finished records itself as abandoned"]
+pub struct Step<'a> {
+    metrics: &'a Metrics,
+    controller: &'static str,
+    step: &'static str,
+    outcome: Option<Outcome>,
+    start: Instant,
+}
+
+impl Step<'_> {
+    /// Records the step as having concluded with `outcome`.
+    pub fn finish(mut self, outcome: Outcome) {
+        self.outcome = Some(outcome);
+    }
+
+    /// Records the step as having concluded with what a reconciler returned:
+    /// [`Outcome::Waiting`] if it asked to be run again, [`Outcome::Applied`]
+    /// if it is done, and [`Outcome::Failed`] if it errored.
+    ///
+    /// A step whose caller holds the `Result` should prefer this over letting
+    /// the error propagate past the guard, since observing the error is what
+    /// separates a failure from an [`Outcome::Abandoned`] pass.
+    pub fn finish_with<E>(self, result: &Result<Option<Action>, E>) {
+        self.finish(Outcome::of_result(result));
+    }
+}
+
+impl Drop for Step<'_> {
+    fn drop(&mut self) {
+        let outcome = self.outcome.unwrap_or(Outcome::Abandoned);
+        self.metrics
+            .steps
+            .with_label_values(&[self.controller, self.step, outcome.as_str()])
+            .inc();
+        self.metrics
+            .step_duration_seconds
+            .with_label_values(&[self.controller, self.step])
+            .observe(self.start.elapsed().as_secs_f64());
+    }
+}
+
+/// Wraps a controller [`Context`](k8s_controller::Context) so that every
+/// reconciliation pass it runs is measured, and so that a failed pass is
+/// published as a Kubernetes warning event on the resource it was reconciling.
+///
+/// This decorates the context rather than sitting beside the controller
+/// because the reconciler's error and the object it was reconciling are only
+/// both in hand here.
+///
+/// NOTE: failures of the finalizer bookkeeping the controller performs around
+/// the reconciler, as opposed to failures of the reconciler itself, are not
+/// covered. They never reach these methods.
+pub struct Observed<Ctx> {
+    inner: Ctx,
+    controller: &'static str,
+    metrics: Arc<Metrics>,
+    publisher: Arc<Publisher>,
+}
+
+impl<Ctx> Observed<Ctx> {
+    /// Wraps `inner`, labelling its metrics and events with `controller`.
+    ///
+    /// `controller` must match the name the reconciler passes to
+    /// [`Metrics::step`], otherwise a pass and its steps land under different
+    /// labels.
+    pub fn new(
+        inner: Ctx,
+        controller: &'static str,
+        metrics: Arc<Metrics>,
+        publisher: Arc<Publisher>,
+    ) -> Self {
+        Self {
+            inner,
+            controller,
+            metrics,
+            publisher,
+        }
+    }
+}
+
+impl<Ctx> Observed<Ctx>
+where
+    Ctx: k8s_controller::Context,
+    <Ctx::Resource as Resource>::DynamicType: Default,
+{
+    /// Publishes a warning event describing `error` on `resource`.
+    async fn publish_failure(
+        &self,
+        entry_point: EntryPoint,
+        resource: &Ctx::Resource,
+        error: &Ctx::Error,
+    ) {
+        self.publisher
+            .publish(
+                resource,
+                Event {
+                    type_: EventType::Warning,
+                    reason: RECONCILIATION_FAILED.into(),
+                    action: entry_point.action().into(),
+                    // The cause chain, not just the outermost message: an error
+                    // like "invalid environment id in license key" is only
+                    // actionable alongside what it was that failed to parse.
+                    note: Some(error.to_string_with_causes()),
+                    secondary: None,
+                },
+            )
+            .await;
+    }
+
+    /// Records what one reconciliation pass did, and reports a failure on the
+    /// resource itself.
+    async fn record(
+        &self,
+        entry_point: EntryPoint,
+        resource: &Ctx::Resource,
+        result: &Result<Option<Action>, Ctx::Error>,
+        start: Instant,
+    ) {
+        self.metrics.record_reconciliation(
+            self.controller,
+            entry_point,
+            Outcome::of_result(result),
+            start.elapsed().as_secs_f64(),
+        );
+        // A pass that succeeded ends whatever the last one reported, so the
+        // next failure files a new event rather than aggregating into one that
+        // has stopped describing the resource.
+        match result {
+            Ok(_) => self.publisher.forget(resource),
+            Err(error) => self.publish_failure(entry_point, resource, error).await,
+        }
+    }
+}
+
+#[async_trait::async_trait]
+impl<Ctx> k8s_controller::Context for Observed<Ctx>
+where
+    Ctx: k8s_controller::Context + Send + Sync + 'static,
+    Ctx::Resource: Send + Sync,
+    Ctx::Error: Send + Sync,
+    <Ctx::Resource as Resource>::DynamicType: Default,
+{
+    type Resource = Ctx::Resource;
+    type Error = Ctx::Error;
+
+    const FINALIZER_NAME: Option<&'static str> = Ctx::FINALIZER_NAME;
+
+    async fn apply(
+        &self,
+        client: Client,
+        resource: &Self::Resource,
+        metadata: &mut TraceMetadata,
+    ) -> Result<Option<Action>, Self::Error> {
+        let start = Instant::now();
+        let result = self.inner.apply(client, resource, metadata).await;
+        self.record(EntryPoint::Apply, resource, &result, start)
+            .await;
+        result
+    }
+
+    async fn cleanup(
+        &self,
+        client: Client,
+        resource: &Self::Resource,
+        metadata: &mut TraceMetadata,
+    ) -> Result<Option<Action>, Self::Error> {
+        let start = Instant::now();
+        let result = self.inner.cleanup(client, resource, metadata).await;
+        self.record(EntryPoint::Cleanup, resource, &result, start)
+            .await;
+        result
+    }
+
+    fn success_action(&self, resource: &Self::Resource) -> Action {
+        self.inner.success_action(resource)
+    }
+
+    // NOTE: `error_action` is deliberately not delegated. Its signature names
+    // `k8s_controller`'s error type, which that crate does not export, so no
+    // context outside the crate can override it. Every context, wrapped or
+    // not, gets the crate's default backoff.
+}
+
+/// Publishes Kubernetes events on the resources a controller reconciles.
+///
+/// Kubernetes expects repeats of an event to aggregate into one object carrying
+/// a count, rather than one object per occurrence. That is what keeps a
+/// reconciliation retrying on a backoff from burying its resource in identical
+/// events, and [`Recorder`] implements it with a cache keyed on everything
+/// about an event except its note, re-sending the cached note on a hit.
+///
+/// Aggregating on that key alone would pin a failing resource's event to
+/// whichever error came first, for as long as the failure lasted. The cache
+/// entry expires six minutes after its last use, but each republish refreshes
+/// it, and the controller's backoff tops out at retrying every 128 to 256
+/// seconds, so the entry of a resource that keeps failing never ages out. An
+/// error that changes along the way, a missing secret first and the real cause
+/// once it appears, would go on reporting the first cause under a count and a
+/// timestamp that both make it look current.
+///
+/// So this aggregates only while the note is unchanged, and files a fresh event
+/// when it changes: one event per distinct message, each counting its own
+/// repeats.
+pub struct Publisher {
+    client: Client,
+    reporter: Reporter,
+    /// The note last published for a resource and reason, with the recorder
+    /// that published it. Sending that same note again through that recorder
+    /// aggregates into the event it already filed; a different note gets a new
+    /// recorder, whose empty cache files a new event instead.
+    ///
+    /// [`Publisher::forget`] bounds this to the resources whose most recent
+    /// reconciliation reported something.
+    published: Mutex<BTreeMap<(String, String), (String, Recorder)>>,
+}
+
+impl Publisher {
+    /// Builds the publisher shared by every controller in this process.
+    ///
+    /// `instance` disambiguates which replica published an event, and should be
+    /// the pod name.
+    pub fn new(client: Client, instance: String) -> Self {
+        Self {
+            client,
+            reporter: Reporter {
+                controller: EVENT_REPORTER.to_owned(),
+                instance: Some(instance),
+            },
+            published: Mutex::new(BTreeMap::new()),
+        }
+    }
+
+    /// Publishes `event` on `resource`, trimming its note to what the API
+    /// server accepts.
+    ///
+    /// Failing to publish is only logged. An event reports something that
+    /// already happened, so failing to file it must not change what the caller
+    /// goes on to do; a missing `events.k8s.io` permission costs visibility,
+    /// not reconciliation.
+    pub async fn publish<K>(&self, resource: &K, mut event: Event)
+    where
+        K: Resource,
+        K::DynamicType: Default,
+    {
+        event.note = event.note.take().map(|note| truncate_note(&note));
+
+        let recorder = self.recorder_for(
+            resource,
+            &event.reason,
+            event.note.clone().unwrap_or_default(),
+        );
+        if let Err(e) = recorder
+            .publish(&event, &resource.object_ref(&Default::default()))
+            .await
+        {
+            warn!(
+                error = %e,
+                reason = %event.reason,
+                kind = %K::kind(&Default::default()),
+                "failed to publish event",
+            );
+        }
+    }
+
+    /// Returns the recorder to publish `note` through: the one that filed the
+    /// resource's last event under `reason` if the note is unchanged, and a
+    /// fresh one otherwise.
+    fn recorder_for<K>(&self, resource: &K, reason: &str, note: String) -> Recorder
+    where
+        K: Resource,
+    {
+        // A resource with no uid has not been persisted, so there is nothing
+        // stable to aggregate against.
+        let Some(uid) = resource.meta().uid.clone() else {
+            return Recorder::new(self.client.clone(), self.reporter.clone());
+        };
+        let key = (uid, reason.to_owned());
+        let mut published = self.published.lock().expect("lock poisoned");
+        if let Some((last_note, recorder)) = published.get(&key)
+            && *last_note == note
+        {
+            return recorder.clone();
+        }
+        let recorder = Recorder::new(self.client.clone(), self.reporter.clone());
+        published.insert(key, (note, recorder.clone()));
+        recorder
+    }
+
+    /// Drops what `resource` published, so that its next event starts afresh
+    /// rather than aggregating into one that has stopped describing it.
+    ///
+    /// Reported when a resource reconciles cleanly, which is what keeps this
+    /// state to the resources currently reporting something.
+    pub fn forget<K>(&self, resource: &K)
+    where
+        K: Resource,
+    {
+        let Some(uid) = resource.meta().uid.as_ref() else {
+            return;
+        };
+        self.published
+            .lock()
+            .expect("lock poisoned")
+            .retain(|(published_uid, _), _| published_uid != uid);
+    }
+}
+
+/// Shortens `note` to fit an event's note, on a character boundary.
+fn truncate_note(note: &str) -> String {
+    if note.len() <= MAX_NOTE_BYTES {
+        return note.to_owned();
+    }
+    const ELLIPSIS: &str = "...";
+    let mut end = MAX_NOTE_BYTES - ELLIPSIS.len();
+    while !note.is_char_boundary(end) {
+        end -= 1;
+    }
+    format!("{}{ELLIPSIS}", &note[..end])
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[mz_ore::test]
+    fn test_truncate_note() {
+        assert_eq!(truncate_note("short"), "short");
+
+        let exact = "a".repeat(MAX_NOTE_BYTES);
+        assert_eq!(truncate_note(&exact), exact);
+
+        let long = "a".repeat(MAX_NOTE_BYTES + 1);
+        let truncated = truncate_note(&long);
+        assert_eq!(truncated.len(), MAX_NOTE_BYTES);
+        assert!(truncated.ends_with("..."));
+
+        // A multi-byte character straddling the cut must not be split.
+        let multibyte = "é".repeat(MAX_NOTE_BYTES);
+        let truncated = truncate_note(&multibyte);
+        assert!(truncated.len() <= MAX_NOTE_BYTES);
+        assert!(truncated.ends_with("..."));
+    }
+}
diff --git a/test/orchestratord/mzcompose.py b/test/orchestratord/mzcompose.py
index 1b5a88b08d58c..7fb5b7d82d879 100644
--- a/test/orchestratord/mzcompose.py
+++ b/test/orchestratord/mzcompose.py
@@ -3340,6 +3340,124 @@ def ca_bundle_refreshed() -> None:
     print("webhook cert rotation test PASSED")
 
 
+def workflow_failure_events(
+    c: Composition,
+    parser: WorkflowArgumentParser,
+) -> None:
+    """Test that a reconciliation failure event reports the current cause.
+
+    Kubernetes aggregates repeats of an event into one object with a count, and
+    `kube`'s recorder keys that aggregation on everything except the event's
+    note. Aggregating a failing resource's events on that key alone pins the
+    note to whichever error came first, and since the controller retries faster
+    than the aggregation window expires, that first note would stand for as
+    long as the resource kept failing.
+
+    So this drives an environment through two *different* failures in a row and
+    requires the event to follow: a Materialize applied before its backend
+    secret exists fails looking the secret up, and once the secret appears with
+    a bad license key it fails validating that instead.
+    """
+    parser.add_argument(
+        "--recreate-cluster",
+        action=argparse.BooleanOptionalAction,
+        help="Recreate cluster if it exists already",
+    )
+    parser.add_argument(
+        "--tag",
+        type=str,
+        help="Custom version tag to use",
+    )
+    parser.add_argument(
+        "--orchestratord-override",
+        default=True,
+        action=argparse.BooleanOptionalAction,
+        help="Override orchestratord tag",
+    )
+    args = parser.parse_args()
+
+    definition = setup(c, args)
+    init(definition)
+
+    # Apply the namespace and the Materialize but not the secret, so the first
+    # failure is the secret lookup. Reconciliation gets this far because
+    # resolving the environment id runs before anything is provisioned.
+    spawn.runv(
+        ["kubectl", "apply", "-f", "-"],
+        stdin=yaml.dump_all(
+            [definition["namespace"], definition["materialize"]]
+        ).encode(),
+    )
+    # The API server's own wording for a missing object, so this anchors that
+    # the first failure really is the secret lookup and not something else.
+    first = wait_for_failure_note(definition, lambda note: "not found" in note.lower())
+    print(f"First failure reported: {first}")
+
+    # Now supply the secret, with a license key that cannot validate, so the
+    # cause changes while the resource keeps failing.
+    secret = copy.deepcopy(definition["secret"])
+    secret["stringData"]["license_key"] = "not-a-real-license-key"
+    spawn.runv(
+        ["kubectl", "apply", "-f", "-"],
+        stdin=yaml.dump_all([secret]).encode(),
+    )
+    # Deliberately not matched against the expected wording: the property under
+    # test is that the reported cause follows the actual one, and asserting the
+    # note merely changed says that without depending on how either error is
+    # phrased.
+    second = wait_for_failure_note(definition, lambda note: note != first)
+    print(f"Second failure reported: {second}")
+    print("workflow_failure_events PASSED")
+
+
+def wait_for_failure_note(
+    definition: dict[str, Any],
+    matches: Callable[[str], bool],
+) -> str:
+    """Wait for a ReconciliationFailed event whose note satisfies `matches`.
+
+    Returns the note, so a caller can compare it against a later one. Times out
+    rather than returning if no note ever matches, which is the failure mode
+    when an event's note is pinned to a cause the resource has moved past.
+    """
+    name = definition["materialize"]["metadata"]["name"]
+    found: list[str] = []
+
+    def check() -> None:
+        events = json.loads(
+            spawn.capture(
+                [
+                    "kubectl",
+                    "get",
+                    # The events.k8s.io API explicitly, rather than the core v1
+                    # view `kubectl get events` serves, which renames these
+                    # fields.
+                    "events.events.k8s.io",
+                    "-n",
+                    "materialize-environment",
+                    "-o",
+                    "json",
+                ],
+            )
+        )["items"]
+        notes = [
+            event.get("note") or ""
+            for event in events
+            if event.get("reason") == "ReconciliationFailed"
+            and event.get("regarding", {}).get("kind") == "Materialize"
+            and event.get("regarding", {}).get("name") == name
+        ]
+        matching = [note for note in notes if matches(note)]
+        assert matching, (
+            f"No ReconciliationFailed event on Materialize/{name} matching the "
+            f"expected cause; notes were {notes}"
+        )
+        found.append(matching[0])
+
+    retry(check, 300)
+    return found[0]
+
+
 def workflow_manually_promote(
     c: Composition,
     parser: WorkflowArgumentParser,
@@ -5064,6 +5182,56 @@ def post_run_check(definition: dict[str, Any], expect_fail: bool) -> None:
         )
         raise ValueError("Never completed")
 
+    if expect_fail:
+        wait_for_reconciliation_failed_event(definition)
+
+
+def wait_for_reconciliation_failed_event(definition: dict[str, Any]) -> None:
+    """Assert that the operator reported the failed reconciliation as an event
+    on the Materialize resource.
+
+    The event is the only place a user without access to the operator's logs
+    sees why their environment is not progressing, so a failure that reaches
+    the logs but not the resource is a regression.
+    """
+    name = definition["materialize"]["metadata"]["name"]
+
+    def check() -> None:
+        events = json.loads(
+            spawn.capture(
+                [
+                    "kubectl",
+                    "get",
+                    # The events.k8s.io API explicitly, rather than the core v1
+                    # view `kubectl get events` serves, which renames the
+                    # fields asserted on below.
+                    "events.events.k8s.io",
+                    "-n",
+                    "materialize-environment",
+                    "-o",
+                    "json",
+                ],
+            )
+        )["items"]
+        matching = [
+            event
+            for event in events
+            if event.get("reason") == "ReconciliationFailed"
+            and event.get("regarding", {}).get("kind") == "Materialize"
+            and event.get("regarding", {}).get("name") == name
+        ]
+        assert matching, (
+            f"Expected a ReconciliationFailed event on Materialize/{name}, found "
+            f"{[(e.get('reason'), e.get('regarding', {}).get('name')) for e in events]}"
+        )
+        event = matching[0]
+        assert event["type"] == "Warning", event
+        assert event["reportingController"] == "orchestratord.materialize.cloud", event
+        # The note carries the error, and is what makes the event actionable.
+        assert event.get("note"), event
+
+    retry(check, 120)
+
 
 def run_balancer(definition: dict[str, Any], expect_fail: bool) -> None:
     defs = [