diff --git a/console-web/app/(dashboard)/page.tsx b/console-web/app/(dashboard)/page.tsx
index a24ce7d..c736186 100755
--- a/console-web/app/(dashboard)/page.tsx
+++ b/console-web/app/(dashboard)/page.tsx
@@ -14,6 +14,7 @@ import { Spinner } from "@/components/ui/spinner"
 import { Table, TableBody, TableCell, TableHead, TableHeader, TableRow } from "@/components/ui/table"
 import * as api from "@/lib/api"
 import { ApiError } from "@/lib/api-client"
+import { normalizeTopologyTenantState } from "@/lib/tenant-state"
 import { cn, formatBinaryBytes, formatK8sMemory } from "@/lib/utils"
 import type { ClusterResourcesResponse, NamespaceItem, NodeInfo } from "@/types/api"
 import type { TopologyOverviewResponse, TopologyTenantState } from "@/types/topology"
@@ -33,6 +34,16 @@ const STATE_THEME: Record<
     dot: "bg-emerald-500",
     card: "border-emerald-200 bg-emerald-50/60",
   },
+  Reconciling: {
+    badge: "border-blue-200 bg-blue-50 text-blue-700",
+    dot: "bg-blue-500",
+    card: "border-blue-200 bg-blue-50/60",
+  },
+  Blocked: {
+    badge: "border-purple-200 bg-purple-50 text-purple-700",
+    dot: "bg-purple-500",
+    card: "border-purple-200 bg-purple-50/60",
+  },
   Updating: {
     badge: "border-blue-200 bg-blue-50 text-blue-700",
     dot: "bg-blue-500",
@@ -60,8 +71,11 @@ function getTreeDotClass(state: string): string {
     case "Ready":
     case "Running":
       return "bg-emerald-500"
+    case "Reconciling":
     case "Updating":
       return "bg-blue-500"
+    case "Blocked":
+      return "bg-purple-500"
     case "Degraded":
     case "Pending":
       return "bg-amber-500"
@@ -151,7 +165,10 @@ export default function DashboardPage() {
   const topologySummary = topology?.cluster.summary
   const tenantCount = topology?.namespaces.reduce((sum, ns) => sum + ns.tenants.length, 0) ?? 0
   const unhealthyCount =
-    topology?.namespaces.reduce((sum, ns) => sum + ns.tenants.filter((t) => t.state !== "Ready").length, 0) ?? 0
+    topology?.namespaces.reduce(
+      (sum, ns) => sum + ns.tenants.filter((tenant) => normalizeTopologyTenantState(tenant.state) !== "Ready").length,
+      0,
+    ) ?? 0
 
   const allPods = topology?.namespaces.flatMap((ns) => ns.tenants.flatMap((t) => t.pods ?? [])) ?? []
   const podTotal = allPods.length
@@ -257,6 +274,7 @@ export default function DashboardPage() {
                                 const tenantId = `t:${ns.name}/${tenant.name}`
                                 const pools = tenant.pools ?? []
                                 const pods = tenant.pods ?? []
+                                const tenantState = normalizeTopologyTenantState(tenant.state)
                                 return (
                                   <div key={tenant.name} className="vtree-vnode">
                                     <button
@@ -264,10 +282,10 @@ export default function DashboardPage() {
                                       className="vtree-vbox vtree-vbox--tenant"
                                       onClick={() => toggleTreeNode(tenantId)}
                                     >
-                                      <span className={cn("vtree-vdot", getTreeDotClass(tenant.state))} />
+                                      <span className={cn("vtree-vdot", getTreeDotClass(tenantState))} />
                                       <span className="vtree-vbox-name">{tenant.name}</span>
-                                      <span className={cn("vtree-vstate", STATE_THEME[tenant.state].badge)}>
-                                        {t(tenant.state)}
+                                      <span className={cn("vtree-vstate", STATE_THEME[tenantState].badge)}>
+                                        {t(tenantState)}
                                       </span>
                                       <RiArrowDownSLine
                                         className={cn("vtree-vchevron", !isTreeExpanded(tenantId) && "rotate-180")}
diff --git a/console-web/app/(dashboard)/tenants/page.tsx b/console-web/app/(dashboard)/tenants/page.tsx
index bc49b66..37ba2d3 100644
--- a/console-web/app/(dashboard)/tenants/page.tsx
+++ b/console-web/app/(dashboard)/tenants/page.tsx
@@ -31,13 +31,16 @@ import { Table, TableBody, TableCell, TableHead, TableHeader, TableRow } from "@
 import * as api from "@/lib/api"
 import { ApiError } from "@/lib/api-client"
 import { routes } from "@/lib/routes"
+import { normalizeTenantLifecycleState } from "@/lib/tenant-state"
 import { parseSizeToBytes, formatBinaryBytes } from "@/lib/utils"
 import type { ServiceInfo, TenantLifecycleState, TenantListItem, TenantStateCountsResponse } from "@/types/api"
 
 const ALL_NAMESPACES = "__all__"
-const TENANT_STATES: TenantLifecycleState[] = ["Ready", "Updating", "Degraded", "NotReady", "Unknown"]
+const TENANT_STATES: TenantLifecycleState[] = ["Ready", "Reconciling", "Blocked", "Degraded", "NotReady", "Unknown"]
 const EMPTY_STATE_COUNTS: Record<TenantLifecycleState, number> = {
   Ready: 0,
+  Reconciling: 0,
+  Blocked: 0,
   Updating: 0,
   Degraded: 0,
   NotReady: 0,
@@ -59,6 +62,18 @@ const STATE_THEME: Record<
     label: "text-emerald-700",
     activeCard: "border-emerald-300 ring-1 ring-emerald-200",
   },
+  Reconciling: {
+    badge: "bg-blue-50 text-blue-700 border-blue-200",
+    dot: "bg-blue-500",
+    label: "text-blue-700",
+    activeCard: "border-blue-300 ring-1 ring-blue-200",
+  },
+  Blocked: {
+    badge: "bg-purple-50 text-purple-700 border-purple-200",
+    dot: "bg-purple-500",
+    label: "text-purple-700",
+    activeCard: "border-purple-300 ring-1 ring-purple-200",
+  },
   Updating: {
     badge: "bg-blue-50 text-blue-700 border-blue-200",
     dot: "bg-blue-500",
@@ -96,18 +111,7 @@ function makeTenantKey(namespace: string, name: string): string {
   return `${namespace}/${name}`
 }
 
-function normalizeTenantState(state: string | null | undefined): TenantLifecycleState {
-  const normalized = (state ?? "")
-    .trim()
-    .toLowerCase()
-    .replace(/[\s_-]/g, "")
-  if (normalized === "ready" || normalized === "running") return "Ready"
-  if (normalized === "updating" || normalized.includes("provision")) return "Updating"
-  if (normalized === "degraded") return "Degraded"
-  if (normalized === "notready" || normalized === "error" || normalized.includes("fail")) return "NotReady"
-  if (normalized === "unknown" || normalized === "stopped") return "Unknown"
-  return "Unknown"
-}
+const normalizeTenantState = normalizeTenantLifecycleState
 
 function parseStateCounts(payload: TenantStateCountsResponse): Record<TenantLifecycleState, number> {
   const result: Record<TenantLifecycleState, number> = { ...EMPTY_STATE_COUNTS }
@@ -392,7 +396,7 @@ export default function TenantsListPage() {
         <p className="text-sm text-muted-foreground">{t("Manage RustFS tenant instances.")}</p>
       </PageHeader>
 
-      <div className="mt-4 grid gap-3 sm:grid-cols-2 xl:grid-cols-6">
+      <div className="mt-4 grid gap-3 sm:grid-cols-2 xl:grid-cols-4 2xl:grid-cols-7">
         <button
           type="button"
           onClick={() => setSelectedState(null)}
diff --git a/console-web/i18n/locales/en-US.json b/console-web/i18n/locales/en-US.json
index 5986281..8ca371b 100755
--- a/console-web/i18n/locales/en-US.json
+++ b/console-web/i18n/locales/en-US.json
@@ -86,6 +86,8 @@
   "Endpoint is unavailable": "Endpoint is unavailable",
   "Copy failed": "Copy failed",
   "Status": "Status",
+  "Reconciling": "Reconciling",
+  "Blocked": "Blocked",
   "Updating": "Updating",
   "Degraded": "Degraded",
   "NotReady": "NotReady",
diff --git a/console-web/i18n/locales/zh-CN.json b/console-web/i18n/locales/zh-CN.json
index b1fa96b..7c269e7 100755
--- a/console-web/i18n/locales/zh-CN.json
+++ b/console-web/i18n/locales/zh-CN.json
@@ -86,6 +86,8 @@
   "Endpoint is unavailable": "Endpoint 不可用",
   "Copy failed": "复制失败",
   "Status": "状态",
+  "Reconciling": "协调中",
+  "Blocked": "阻塞",
   "Updating": "更新中",
   "Degraded": "降级",
   "NotReady": "未就绪",
diff --git a/console-web/lib/tenant-state.ts b/console-web/lib/tenant-state.ts
new file mode 100644
index 0000000..3294c15
--- /dev/null
+++ b/console-web/lib/tenant-state.ts
@@ -0,0 +1,34 @@
+import type { TenantLifecycleState } from "@/types/api"
+import type { TopologyTenantState } from "@/types/topology"
+
+function normalizeStateToken(state: string | null | undefined): string {
+  return (state ?? "")
+    .trim()
+    .toLowerCase()
+    .replace(/[\s_-]/g, "")
+}
+
+export function normalizeTenantLifecycleState(state: string | null | undefined): TenantLifecycleState {
+  const normalized = normalizeStateToken(state)
+
+  if (normalized === "ready" || normalized === "running") return "Ready"
+  if (
+    normalized === "reconciling" ||
+    normalized === "updating" ||
+    normalized === "creating" ||
+    normalized.includes("provision")
+  ) {
+    return "Reconciling"
+  }
+  if (normalized === "blocked") return "Blocked"
+  if (normalized === "degraded") return "Degraded"
+  if (normalized === "notready" || normalized === "error" || normalized.includes("fail")) {
+    return "NotReady"
+  }
+  if (normalized === "unknown" || normalized === "stopped") return "Unknown"
+  return "Unknown"
+}
+
+export function normalizeTopologyTenantState(state: string | null | undefined): TopologyTenantState {
+  return normalizeTenantLifecycleState(state)
+}
diff --git a/console-web/types/api.ts b/console-web/types/api.ts
index 558681c..a7f82fb 100644
--- a/console-web/types/api.ts
+++ b/console-web/types/api.ts
@@ -12,6 +12,13 @@ export interface TenantListItem {
   namespace: string
   pools: PoolInfo[]
   state: string
+  ready?: boolean
+  reconciling?: boolean
+  degraded?: boolean
+  primary_reason?: string | null
+  generation?: number | null
+  observed_generation?: number | null
+  stale?: boolean
   created_at: string | null
 }
 
@@ -19,7 +26,14 @@ export interface TenantListResponse {
   tenants: TenantListItem[]
 }
 
-export type TenantLifecycleState = "Ready" | "Updating" | "Degraded" | "NotReady" | "Unknown"
+export type TenantLifecycleState =
+  | "Ready"
+  | "Reconciling"
+  | "Blocked"
+  | "Updating"
+  | "Degraded"
+  | "NotReady"
+  | "Unknown"
 
 export interface TenantStateCountItem {
   state: string
@@ -46,11 +60,35 @@ export interface ServiceInfo {
   ports: ServicePort[]
 }
 
+export interface TenantCondition {
+  type: string
+  status: string
+  reason: string
+  message: string
+  last_transition_time: string | null
+  observed_generation: number | null
+}
+
+export interface TenantStatusSummary {
+  current_state: string
+  ready: boolean
+  reconciling: boolean
+  degraded: boolean
+  primary_reason: string | null
+  primary_message: string | null
+  observed_generation: number | null
+  stale: boolean
+  next_actions: string[]
+}
+
 export interface TenantDetailsResponse {
   name: string
   namespace: string
   pools: PoolInfo[]
   state: string
+  status_summary: TenantStatusSummary
+  conditions: TenantCondition[]
+  next_actions: string[]
   image: string | null
   mount_path: string | null
   created_at: string | null
diff --git a/console-web/types/topology.ts b/console-web/types/topology.ts
index 5533906..1ced063 100644
--- a/console-web/types/topology.ts
+++ b/console-web/types/topology.ts
@@ -1,4 +1,4 @@
-export type TopologyTenantState = "Ready" | "Updating" | "Degraded" | "NotReady" | "Unknown"
+export type TopologyTenantState = "Ready" | "Reconciling" | "Blocked" | "Updating" | "Degraded" | "NotReady" | "Unknown"
 
 export interface TopologyClusterSummary {
   nodes: number
@@ -48,6 +48,11 @@ export interface TopologyTenant {
   name: string
   namespace: string
   state: TopologyTenantState
+  ready?: boolean
+  reconciling?: boolean
+  degraded?: boolean
+  stale?: boolean
+  primary_reason?: string | null
   created_at: string | null
   summary: TopologyTenantSummary
   pools?: TopologyPool[]
diff --git a/src/console/handlers/tenants.rs b/src/console/handlers/tenants.rs
index b9d61ca..261d129 100755
--- a/src/console/handlers/tenants.rs
+++ b/src/console/handlers/tenants.rs
@@ -160,6 +160,10 @@ pub async fn get_tenant_details(
         })
         .collect();
 
+    let status_summary = tenant_status_summary(&tenant);
+    let conditions = tenant_conditions(&tenant);
+    let next_actions = status_summary.next_actions.clone();
+
     Ok(Json(TenantDetailsResponse {
         name: tenant.name_any(),
         namespace: tenant.namespace().unwrap_or_default(),
@@ -173,11 +177,10 @@ pub async fn get_tenant_details(
                 volumes_per_server: p.persistence.volumes_per_server,
             })
             .collect(),
-        state: tenant
-            .status
-            .as_ref()
-            .map(|s| s.current_state.to_string())
-            .unwrap_or_else(|| "Unknown".to_string()),
+        state: status_summary.current_state.clone(),
+        status_summary,
+        conditions,
+        next_actions,
         image: tenant.spec.image.clone(),
         mount_path: tenant.spec.mount_path.clone(),
         created_at: tenant
@@ -291,25 +294,9 @@ pub async fn create_tenant(
         .await
         .map_err(|e| error::map_kube_error(e, format!("Tenant '{}'", req.name)))?;
 
-    Ok(Json(TenantListItem {
-        name: created.name_any(),
-        namespace: created.namespace().unwrap_or_default(),
-        pools: created
-            .spec
-            .pools
-            .iter()
-            .map(|p| PoolInfo {
-                name: p.name.clone(),
-                servers: p.servers,
-                volumes_per_server: p.persistence.volumes_per_server,
-            })
-            .collect(),
-        state: "Creating".to_string(),
-        created_at: created
-            .metadata
-            .creation_timestamp
-            .map(|ts| ts.0.to_rfc3339()),
-    }))
+    let item = tenant_to_list_item(created);
+
+    Ok(Json(item))
 }
 
 /// Delete a Tenant CR.
@@ -458,29 +445,7 @@ pub async fn update_tenant(
     Ok(Json(UpdateTenantResponse {
         success: true,
         message: format!("Tenant updated: {}", updated_fields.join(", ")),
-        tenant: TenantListItem {
-            name: updated_tenant.name_any(),
-            namespace: updated_tenant.namespace().unwrap_or_default(),
-            pools: updated_tenant
-                .spec
-                .pools
-                .iter()
-                .map(|p| PoolInfo {
-                    name: p.name.clone(),
-                    servers: p.servers,
-                    volumes_per_server: p.persistence.volumes_per_server,
-                })
-                .collect(),
-            state: updated_tenant
-                .status
-                .as_ref()
-                .map(|s| s.current_state.to_string())
-                .unwrap_or_else(|| "Unknown".to_string()),
-            created_at: updated_tenant
-                .metadata
-                .creation_timestamp
-                .map(|ts| ts.0.to_rfc3339()),
-        },
+        tenant: tenant_to_list_item(updated_tenant),
     }))
 }
 
@@ -623,36 +588,15 @@ fn build_tenant_list_items(
         .collect()
 }
 
-fn tenant_to_list_item(t: Tenant) -> TenantListItem {
-    let state = tenant_state(&t);
-    TenantListItem {
-        name: t.name_any(),
-        namespace: t.namespace().unwrap_or_default(),
-        pools: t
-            .spec
-            .pools
-            .iter()
-            .map(|p| PoolInfo {
-                name: p.name.clone(),
-                servers: p.servers,
-                volumes_per_server: p.persistence.volumes_per_server,
-            })
-            .collect(),
-        state,
-        created_at: t.metadata.creation_timestamp.map(|ts| ts.0.to_rfc3339()),
-    }
-}
-
 fn tenant_state(t: &Tenant) -> String {
-    t.status
-        .as_ref()
-        .map(|s| s.current_state.to_string())
-        .unwrap_or_else(|| "Unknown".to_string())
+    tenant_status_summary(t).current_state
 }
 
 fn state_matches_filter(state: &str, state_filter: Option<&str>) -> bool {
     match state_filter {
-        Some(filter) => state.eq_ignore_ascii_case(filter),
+        Some(filter) => canonical_console_state_filter(Some(filter)).is_some_and(|filter| {
+            canonical_console_state(Some(state)).eq_ignore_ascii_case(&filter)
+        }),
         None => true,
     }
 }
@@ -669,3 +613,20 @@ fn summarize_tenant_states(tenants: &[Tenant]) -> TenantStateCountsResponse {
         counts,
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::state_matches_filter;
+
+    #[test]
+    fn state_filter_is_case_insensitive_for_known_states() {
+        assert!(state_matches_filter("Ready", Some("ready")));
+        assert!(state_matches_filter("Reconciling", Some("updating")));
+        assert!(state_matches_filter("Blocked", Some("blocked")));
+    }
+
+    #[test]
+    fn unknown_filter_value_does_not_match_unknown_state() {
+        assert!(!state_matches_filter("Unknown", Some("foo")));
+    }
+}
diff --git a/src/console/handlers/topology.rs b/src/console/handlers/topology.rs
index a67b413..277f6c7 100644
--- a/src/console/handlers/topology.rs
+++ b/src/console/handlers/topology.rs
@@ -18,6 +18,7 @@ use crate::console::{
         format_cpu_from_millicores, format_memory_from_bytes, parse_cpu_to_millicores,
         parse_memory_to_bytes,
     },
+    models::tenant::tenant_status_summary,
     models::topology::*,
     state::Claims,
 };
@@ -199,13 +200,14 @@ pub async fn get_topology_overview(
                 .map(|t| {
                     let name = t.name_any();
                     let namespace = t.namespace().unwrap_or_default();
-                    let state = t
-                        .status
-                        .as_ref()
-                        .map(|s| s.current_state.clone())
-                        .unwrap_or_else(|| "Unknown".to_string());
-
-                    if !is_healthy_state(&state) {
+                    let status_summary = tenant_status_summary(t);
+                    let state = status_summary.current_state.clone();
+
+                    if !(status_summary.ready
+                        && !status_summary.reconciling
+                        && !status_summary.degraded
+                        && !status_summary.stale)
+                    {
                         unhealthy_count += 1;
                     }
 
@@ -276,6 +278,11 @@ pub async fn get_topology_overview(
                         name,
                         namespace,
                         state,
+                        ready: status_summary.ready,
+                        reconciling: status_summary.reconciling,
+                        degraded: status_summary.degraded,
+                        stale: status_summary.stale,
+                        primary_reason: status_summary.primary_reason,
                         created_at,
                         summary: TopologyTenantSummary {
                             pool_count,
@@ -326,11 +333,6 @@ pub async fn get_topology_overview(
     }))
 }
 
-/// Whether the tenant aggregate state counts as healthy for the UI.
-fn is_healthy_state(state: &str) -> bool {
-    matches!(state, "Ready" | "Initialized")
-}
-
 /// Map operator `PoolState` to a short UI label.
 fn map_pool_state(state: &PoolState) -> String {
     match state {
diff --git a/src/console/models/tenant.rs b/src/console/models/tenant.rs
index f8b1131..16d017d 100755
--- a/src/console/models/tenant.rs
+++ b/src/console/models/tenant.rs
@@ -12,6 +12,14 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use crate::types::v1alpha1::{
+    status::{
+        ConditionStatus, ConditionType, CurrentState, Reason, Status, canonical_filter_state,
+        canonical_state, next_actions_for_reason, primary_condition, summarize_current_state,
+    },
+    tenant::Tenant,
+};
+use kube::ResourceExt;
 use serde::{Deserialize, Serialize};
 use utoipa::ToSchema;
 
@@ -22,6 +30,13 @@ pub struct TenantListItem {
     pub namespace: String,
     pub pools: Vec<PoolInfo>,
     pub state: String,
+    pub ready: bool,
+    pub reconciling: bool,
+    pub degraded: bool,
+    pub primary_reason: Option<String>,
+    pub generation: Option<i64>,
+    pub observed_generation: Option<i64>,
+    pub stale: bool,
     pub created_at: Option<String>,
 }
 
@@ -51,7 +66,7 @@ pub struct TenantListQuery {
 pub struct TenantStateCountsResponse {
     /// Total number of tenants
     pub total: u32,
-    /// Counts keyed by state, e.g. Ready/Updating/Degraded/NotReady/Unknown
+    /// Counts keyed by state, e.g. Ready/Reconciling/Blocked/Degraded/NotReady/Unknown
     pub counts: std::collections::BTreeMap<String, u32>,
 }
 
@@ -62,12 +77,39 @@ pub struct TenantDetailsResponse {
     pub namespace: String,
     pub pools: Vec<PoolInfo>,
     pub state: String,
+    pub status_summary: TenantStatusSummary,
+    pub conditions: Vec<TenantCondition>,
+    pub next_actions: Vec<String>,
     pub image: Option<String>,
     pub mount_path: Option<String>,
     pub created_at: Option<String>,
     pub services: Vec<ServiceInfo>,
 }
 
+#[derive(Debug, Serialize, ToSchema, Clone)]
+pub struct TenantCondition {
+    #[serde(rename = "type")]
+    pub type_: String,
+    pub status: String,
+    pub reason: String,
+    pub message: String,
+    pub last_transition_time: Option<String>,
+    pub observed_generation: Option<i64>,
+}
+
+#[derive(Debug, Serialize, ToSchema, Clone)]
+pub struct TenantStatusSummary {
+    pub current_state: String,
+    pub ready: bool,
+    pub reconciling: bool,
+    pub degraded: bool,
+    pub primary_reason: Option<String>,
+    pub primary_message: Option<String>,
+    pub observed_generation: Option<i64>,
+    pub stale: bool,
+    pub next_actions: Vec<String>,
+}
+
 /// Exposed Service summary
 #[derive(Debug, Serialize, ToSchema)]
 pub struct ServiceInfo {
@@ -179,3 +221,329 @@ pub struct UpdateTenantResponse {
 pub struct TenantYAML {
     pub yaml: String,
 }
+
+pub fn tenant_status_summary(tenant: &Tenant) -> TenantStatusSummary {
+    let status = tenant.status.as_ref();
+    let generation = tenant.metadata.generation;
+    let observed_generation = status.and_then(|status| status.observed_generation);
+    let stale = match (observed_generation, generation) {
+        (Some(observed), Some(generation)) => observed < generation,
+        (None, Some(_)) => true,
+        _ => false,
+    };
+
+    let has_conditions = status.is_some_and(|status| !status.conditions.is_empty());
+    let derived_state = status
+        .map(|status| {
+            if has_conditions {
+                summarize_current_state(status)
+            } else {
+                status_current_state(status)
+            }
+        })
+        .unwrap_or_else(|| CurrentState::Unknown.as_str().to_string());
+
+    let mut current_state = derived_state;
+    let mut ready = if has_conditions {
+        condition_is_true(status, ConditionType::Ready.as_str())
+    } else {
+        legacy_ready_for_state(&current_state)
+    };
+    let mut reconciling = if has_conditions {
+        condition_is_true(status, ConditionType::Reconciling.as_str())
+            || condition_is_true(status, "Progressing")
+    } else {
+        legacy_reconciling_for_state(&current_state)
+    };
+    let degraded = if has_conditions {
+        condition_is_true(status, ConditionType::Degraded.as_str())
+    } else {
+        legacy_degraded_for_state(&current_state)
+    };
+
+    let primary = status.and_then(primary_condition);
+    let mut primary_reason = primary.map(|condition| condition.reason.clone());
+    let mut primary_message = primary.map(|condition| condition.message.clone());
+
+    if stale {
+        current_state = CurrentState::Reconciling.as_str().to_string();
+        ready = false;
+        reconciling = true;
+        primary_reason = Some(Reason::ObservedGenerationStale.as_str().to_string());
+        primary_message =
+            Some("Tenant spec changed and has not been observed by the operator yet".to_string());
+    }
+
+    let next_actions = primary_reason
+        .as_deref()
+        .map(next_actions_for_reason)
+        .unwrap_or_default()
+        .into_iter()
+        .map(ToOwned::to_owned)
+        .collect();
+
+    TenantStatusSummary {
+        current_state,
+        ready,
+        reconciling,
+        degraded,
+        primary_reason,
+        primary_message,
+        observed_generation,
+        stale,
+        next_actions,
+    }
+}
+
+pub fn tenant_conditions(tenant: &Tenant) -> Vec<TenantCondition> {
+    tenant
+        .status
+        .as_ref()
+        .map(|status| {
+            status
+                .conditions
+                .iter()
+                .map(|condition| TenantCondition {
+                    type_: condition.type_.clone(),
+                    status: condition.status.clone(),
+                    reason: condition.reason.clone(),
+                    message: condition.message.clone(),
+                    last_transition_time: condition.last_transition_time.clone(),
+                    observed_generation: condition.observed_generation,
+                })
+                .collect()
+        })
+        .unwrap_or_default()
+}
+
+pub fn tenant_to_list_item(tenant: Tenant) -> TenantListItem {
+    let summary = tenant_status_summary(&tenant);
+    TenantListItem {
+        name: tenant.name_any(),
+        namespace: tenant.namespace().unwrap_or_default(),
+        pools: tenant
+            .spec
+            .pools
+            .iter()
+            .map(|p| PoolInfo {
+                name: p.name.clone(),
+                servers: p.servers,
+                volumes_per_server: p.persistence.volumes_per_server,
+            })
+            .collect(),
+        state: summary.current_state,
+        ready: summary.ready,
+        reconciling: summary.reconciling,
+        degraded: summary.degraded,
+        primary_reason: summary.primary_reason,
+        generation: tenant.metadata.generation,
+        observed_generation: summary.observed_generation,
+        stale: summary.stale,
+        created_at: tenant
+            .metadata
+            .creation_timestamp
+            .map(|ts| ts.0.to_rfc3339()),
+    }
+}
+
+pub fn canonical_console_state(state: Option<&str>) -> String {
+    canonical_state(state)
+}
+
+pub fn canonical_console_state_filter(state: Option<&str>) -> Option<String> {
+    canonical_filter_state(state)
+}
+
+fn status_current_state(status: &Status) -> String {
+    if status.current_state.is_empty() {
+        summarize_current_state(status)
+    } else {
+        canonical_console_state(Some(&status.current_state))
+    }
+}
+
+fn legacy_ready_for_state(state: &str) -> bool {
+    canonical_console_state(Some(state)) == CurrentState::Ready.as_str()
+}
+
+fn legacy_reconciling_for_state(state: &str) -> bool {
+    canonical_console_state(Some(state)) == CurrentState::Reconciling.as_str()
+}
+
+fn legacy_degraded_for_state(state: &str) -> bool {
+    matches!(
+        canonical_console_state(Some(state)).as_str(),
+        "Blocked" | "Degraded"
+    )
+}
+
+fn condition_is_true(status: Option<&Status>, type_: &str) -> bool {
+    status.is_some_and(|status| {
+        status
+            .condition_by_type(type_)
+            .is_some_and(|condition| condition.status == ConditionStatus::True.as_str())
+    })
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::types::v1alpha1::status::{Condition, Status};
+
+    #[test]
+    fn tenant_summary_prefers_blocked_reason_and_actions() {
+        let mut tenant = crate::tests::create_test_tenant(None, None);
+        tenant.metadata.generation = Some(7);
+        tenant.status = Some(Status {
+            current_state: "Blocked".to_string(),
+            available_replicas: 0,
+            pools: Vec::new(),
+            observed_generation: Some(7),
+            conditions: vec![
+                condition("Reconciling", "True", "RolloutInProgress"),
+                condition("CredentialsReady", "False", "CredentialSecretNotFound"),
+                condition("Degraded", "True", "CredentialSecretNotFound"),
+            ],
+        });
+
+        let summary = tenant_status_summary(&tenant);
+
+        assert_eq!(summary.current_state, "Blocked");
+        assert_eq!(
+            summary.primary_reason.as_deref(),
+            Some("CredentialSecretNotFound")
+        );
+        assert_eq!(summary.next_actions, vec!["createCredentialSecret"]);
+        assert!(!summary.stale);
+    }
+
+    #[test]
+    fn tenant_summary_detects_stale_generation_as_reconciling() {
+        let mut tenant = crate::tests::create_test_tenant(None, None);
+        tenant.metadata.generation = Some(8);
+        tenant.status = Some(Status {
+            current_state: "Ready".to_string(),
+            available_replicas: 4,
+            pools: Vec::new(),
+            observed_generation: Some(7),
+            conditions: vec![
+                condition("Ready", "True", "ReconcileSucceeded"),
+                condition("Degraded", "False", "ReconcileSucceeded"),
+            ],
+        });
+
+        let summary = tenant_status_summary(&tenant);
+
+        assert_eq!(summary.current_state, "Reconciling");
+        assert!(!summary.ready);
+        assert!(summary.reconciling);
+        assert!(summary.stale);
+        assert_eq!(
+            summary.primary_reason.as_deref(),
+            Some("ObservedGenerationStale")
+        );
+        assert_eq!(summary.next_actions, vec!["waitForReconcile"]);
+    }
+
+    #[test]
+    fn tenant_summary_treats_missing_observed_generation_as_stale() {
+        let mut tenant = crate::tests::create_test_tenant(None, None);
+        tenant.metadata.generation = Some(9);
+        tenant.status = Some(Status {
+            current_state: "Ready".to_string(),
+            available_replicas: 4,
+            pools: Vec::new(),
+            observed_generation: None,
+            conditions: vec![condition("Ready", "True", "ReconcileSucceeded")],
+        });
+
+        let summary = tenant_status_summary(&tenant);
+
+        assert_eq!(summary.current_state, "Reconciling");
+        assert!(summary.stale);
+        assert!(summary.reconciling);
+        assert!(!summary.ready);
+        assert_eq!(
+            summary.primary_reason.as_deref(),
+            Some("ObservedGenerationStale")
+        );
+        assert_eq!(summary.next_actions, vec!["waitForReconcile"]);
+    }
+
+    #[test]
+    fn tenant_summary_treats_missing_status_as_stale() {
+        let mut tenant = crate::tests::create_test_tenant(None, None);
+        tenant.metadata.generation = Some(2);
+        tenant.status = None;
+
+        let summary = tenant_status_summary(&tenant);
+
+        assert_eq!(summary.current_state, "Reconciling");
+        assert!(summary.stale);
+        assert!(summary.reconciling);
+        assert!(!summary.ready);
+        assert_eq!(
+            summary.primary_reason.as_deref(),
+            Some("ObservedGenerationStale")
+        );
+    }
+
+    #[test]
+    fn tenant_summary_uses_legacy_current_state_when_conditions_are_missing() {
+        let mut tenant = crate::tests::create_test_tenant(None, None);
+        tenant.metadata.generation = Some(3);
+        tenant.status = Some(Status {
+            current_state: "Ready".to_string(),
+            observed_generation: Some(3),
+            ..Default::default()
+        });
+
+        let summary = tenant_status_summary(&tenant);
+
+        assert_eq!(summary.current_state, "Ready");
+        assert!(summary.ready);
+        assert!(!summary.reconciling);
+        assert!(!summary.degraded);
+    }
+
+    #[test]
+    fn tenant_summary_prefers_conditions_over_legacy_current_state() {
+        let mut tenant = crate::tests::create_test_tenant(None, None);
+        tenant.metadata.generation = Some(3);
+        tenant.status = Some(Status {
+            current_state: "Ready".to_string(),
+            observed_generation: Some(3),
+            conditions: vec![condition("Reconciling", "True", "RolloutInProgress")],
+            ..Default::default()
+        });
+
+        let summary = tenant_status_summary(&tenant);
+
+        assert_eq!(summary.current_state, "Reconciling");
+        assert!(!summary.ready);
+        assert!(summary.reconciling);
+    }
+
+    #[test]
+    fn canonical_console_state_accepts_legacy_updating() {
+        assert_eq!(
+            canonical_console_state(Some("Updating")),
+            "Reconciling".to_string()
+        );
+        assert_eq!(
+            canonical_console_state(Some("Initialized")),
+            "Ready".to_string()
+        );
+    }
+
+    fn condition(type_: &str, status: &str, reason: &str) -> Condition {
+        Condition {
+            type_: type_.to_string(),
+            status: status.to_string(),
+            last_transition_time: Some("now".to_string()),
+            observed_generation: None,
+            reason: reason.to_string(),
+            message: reason.to_string(),
+        }
+    }
+}
diff --git a/src/console/models/topology.rs b/src/console/models/topology.rs
index bffa49a..7f629e0 100644
--- a/src/console/models/topology.rs
+++ b/src/console/models/topology.rs
@@ -60,6 +60,11 @@ pub struct TopologyTenant {
     pub name: String,
     pub namespace: String,
     pub state: String,
+    pub ready: bool,
+    pub reconciling: bool,
+    pub degraded: bool,
+    pub stale: bool,
+    pub primary_reason: Option<String>,
     pub created_at: Option<String>,
     pub summary: TopologyTenantSummary,
     #[serde(skip_serializing_if = "Option::is_none")]
diff --git a/src/console/openapi.rs b/src/console/openapi.rs
index 3aaf347..c78ed2c 100644
--- a/src/console/openapi.rs
+++ b/src/console/openapi.rs
@@ -36,9 +36,9 @@ use crate::console::models::pool::{
 };
 use crate::console::models::tenant::{
     CreatePoolRequest, CreateTenantRequest, DeleteTenantResponse, EnvVar, LoggingConfig, PoolInfo,
-    ServiceInfo, ServicePort, TenantDetailsResponse, TenantListItem, TenantListQuery,
-    TenantListResponse, TenantStateCountsResponse, TenantYAML, UpdateTenantRequest,
-    UpdateTenantResponse,
+    ServiceInfo, ServicePort, TenantCondition, TenantDetailsResponse, TenantListItem,
+    TenantListQuery, TenantListResponse, TenantStateCountsResponse, TenantStatusSummary,
+    TenantYAML, UpdateTenantRequest, UpdateTenantResponse,
 };
 use crate::console::models::topology::{
     TopologyCluster, TopologyClusterSummary, TopologyNamespace, TopologyNode,
@@ -84,6 +84,8 @@ use crate::console::models::topology::{
         TenantListResponse,
         TenantListQuery,
         TenantStateCountsResponse,
+        TenantCondition,
+        TenantStatusSummary,
         TenantDetailsResponse,
         CreateTenantRequest,
         CreatePoolRequest,
diff --git a/src/context.rs b/src/context.rs
index 2f475a1..e76e5d6 100755
--- a/src/context.rs
+++ b/src/context.rs
@@ -100,6 +100,57 @@ fn validate_local_kms_tenant(
     Ok(())
 }
 
+fn status_semantically_equal(
+    current: Option<&types::v1alpha1::status::Status>,
+    next: &types::v1alpha1::status::Status,
+) -> bool {
+    let Some(current) = current else {
+        return false;
+    };
+
+    let mut current = current.clone();
+    let mut next = next.clone();
+    normalize_status_for_compare(&mut current);
+    normalize_status_for_compare(&mut next);
+    current == next
+}
+
+fn normalize_status_for_compare(status: &mut types::v1alpha1::status::Status) {
+    for pool in &mut status.pools {
+        pool.last_update_time = None;
+    }
+}
+
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+pub(crate) enum SecretValidationKind {
+    Credential,
+    Kms,
+}
+
+pub(crate) fn is_kube_not_found(error: &Error) -> bool {
+    matches!(
+        error,
+        Error::Kube {
+            source: kube::Error::Api(response),
+        } if response.code == 404
+    )
+}
+
+pub(crate) fn map_secret_get_error(
+    error: Error,
+    name: String,
+    kind: SecretValidationKind,
+) -> Error {
+    if !is_kube_not_found(&error) {
+        return error;
+    }
+
+    match kind {
+        SecretValidationKind::Credential => Error::CredentialSecretNotFound { name },
+        SecretValidationKind::Kms => Error::KmsSecretNotFound { name },
+    }
+}
+
 pub struct Context {
     pub(crate) client: kube::Client,
     pub(crate) recorder: Recorder,
@@ -177,6 +228,18 @@ impl Context {
             .await
     }
 
+    pub async fn patch_status_if_changed(
+        &self,
+        resource: &Tenant,
+        status: crate::types::v1alpha1::status::Status,
+    ) -> Result<Option<Tenant>, Error> {
+        if status_semantically_equal(resource.status.as_ref(), &status) {
+            return Ok(None);
+        }
+
+        self.update_status(resource, status).await.map(Some)
+    }
+
     pub async fn delete<T>(&self, name: &str, namespace: &str) -> Result<(), Error>
     where
         T: Resource<Scope = NamespaceResourceScope> + Clone + DeserializeOwned + Debug,
@@ -218,6 +281,19 @@ impl Context {
         api.list(&ListParams::default()).context(KubeSnafu).await
     }
 
+    pub async fn list_with_params<T>(
+        &self,
+        namespace: &str,
+        params: &ListParams,
+    ) -> Result<ObjectList<T>, Error>
+    where
+        T: Clone + DeserializeOwned + Debug + Resource<Scope = NamespaceResourceScope>,
+        <T as kube::Resource>::DynamicType: Default,
+    {
+        let api: Api<T> = Api::namespaced(self.client.clone(), namespace);
+        api.list(params).context(KubeSnafu).await
+    }
+
     pub async fn apply<T>(&self, resource: &T, namespace: &str) -> Result<T, Error>
     where
         T: Clone + Serialize + DeserializeOwned + Debug + Resource<Scope = NamespaceResourceScope>,
@@ -258,12 +334,16 @@ impl Context {
         if let Some(ref cfg) = tenant.spec.creds_secret
             && !cfg.name.is_empty()
         {
-            let secret: Secret = self
-                .get(&cfg.name, &tenant.namespace()?)
-                .await
-                .map_err(|_| Error::CredentialSecretNotFound {
-                    name: cfg.name.clone(),
-                })?;
+            let secret: Secret = match self.get(&cfg.name, &tenant.namespace()?).await {
+                Ok(secret) => secret,
+                Err(error) => {
+                    return Err(map_secret_get_error(
+                        error,
+                        cfg.name.clone(),
+                        SecretValidationKind::Credential,
+                    ));
+                }
+            };
 
             // Validate Secret has required keys
             if let Some(data) = secret.data {
@@ -380,12 +460,16 @@ impl Context {
             return Ok(());
         }
 
-        let secret: Secret = self
-            .get(&secret_ref.name, &tenant.namespace()?)
-            .await
-            .map_err(|_| Error::KmsSecretNotFound {
-                name: secret_ref.name.clone(),
-            })?;
+        let secret: Secret = match self.get(&secret_ref.name, &tenant.namespace()?).await {
+            Ok(secret) => secret,
+            Err(error) => {
+                return Err(map_secret_get_error(
+                    error,
+                    secret_ref.name.clone(),
+                    SecretValidationKind::Kms,
+                ));
+            }
+        };
 
         if enc.backend == KmsBackendType::Vault {
             let has_token = secret
@@ -490,6 +574,7 @@ impl Context {
 mod validate_local_kms_tests {
     use super::Error;
     use super::validate_local_kms_tenant;
+    use super::{SecretValidationKind, map_secret_get_error};
     use crate::types::v1alpha1::encryption::LocalKmsConfig;
     use crate::types::v1alpha1::persistence::PersistenceConfig;
     use crate::types::v1alpha1::pool::Pool;
@@ -506,6 +591,52 @@ mod validate_local_kms_tests {
         }
     }
 
+    fn api_error(code: u16, reason: &str) -> Error {
+        Error::Kube {
+            source: kube::Error::Api(kube::error::ErrorResponse {
+                status: "Failure".to_string(),
+                message: reason.to_string(),
+                reason: reason.to_string(),
+                code,
+            }),
+        }
+    }
+
+    #[test]
+    fn credential_secret_get_maps_only_404_to_not_found() {
+        let err = map_secret_get_error(
+            api_error(404, "NotFound"),
+            "creds".to_string(),
+            SecretValidationKind::Credential,
+        );
+
+        assert!(matches!(err, Error::CredentialSecretNotFound { name } if name == "creds"));
+    }
+
+    #[test]
+    fn credential_secret_get_preserves_forbidden_as_kube_error() {
+        let err = map_secret_get_error(
+            api_error(403, "Forbidden"),
+            "creds".to_string(),
+            SecretValidationKind::Credential,
+        );
+
+        assert!(
+            matches!(err, Error::Kube { source: kube::Error::Api(response) } if response.code == 403)
+        );
+    }
+
+    #[test]
+    fn kms_secret_get_maps_only_404_to_not_found() {
+        let err = map_secret_get_error(
+            api_error(404, "NotFound"),
+            "kms".to_string(),
+            SecretValidationKind::Kms,
+        );
+
+        assert!(matches!(err, Error::KmsSecretNotFound { name } if name == "kms"));
+    }
+
     #[test]
     fn local_kms_default_key_dir_ok_single_replica() {
         validate_local_kms_tenant(None, &[pool(1)]).unwrap();
diff --git a/src/lib.rs b/src/lib.rs
index 1f5315f..79bd2f1 100755
--- a/src/lib.rs
+++ b/src/lib.rs
@@ -30,6 +30,7 @@ use tracing::{info, warn};
 
 mod context;
 pub mod reconcile;
+mod status;
 pub mod types;
 pub mod utils;
 
diff --git a/src/reconcile.rs b/src/reconcile.rs
index 529e524..61d44ac 100755
--- a/src/reconcile.rs
+++ b/src/reconcile.rs
@@ -13,6 +13,8 @@
 // limitations under the License.
 
 use crate::context::Context;
+use crate::status::{StatusBuilder, StatusError};
+use crate::types::v1alpha1::status::{ConditionType, Reason, Status};
 use crate::types::v1alpha1::tenant::Tenant;
 use crate::{context, types};
 use k8s_openapi::api::core::v1 as corev1;
@@ -23,7 +25,15 @@ use kube::runtime::events::EventType;
 use snafu::Snafu;
 use std::sync::Arc;
 use std::time::Duration;
-use tracing::{debug, error, warn};
+use tracing::{debug, error, info, warn};
+
+mod phases;
+
+use phases::{
+    finalize_tenant_status, maybe_cleanup_terminating_pods, reconcile_pool_statefulsets,
+    reconcile_rbac_resources, reconcile_services, validate_no_pool_rename,
+    validate_tenant_prerequisites,
+};
 
 #[derive(Snafu, Debug)]
 pub enum Error {
@@ -47,384 +57,267 @@ pub async fn reconcile_rustfs(tenant: Arc<Tenant>, ctx: Arc<Context>) -> Result<
         return Ok(Action::await_change());
     }
 
-    // Validate tenant name is DNS-1035 compliant (required for derived Service names)
-    if let Err(e) = latest_tenant.validate_name() {
-        let _ = ctx
-            .record(
-                &latest_tenant,
-                EventType::Warning,
-                "InvalidTenantName",
-                &format!("{}", e),
-            )
-            .await;
-        return Err(e.into());
+    if should_mark_reconcile_started(&latest_tenant) {
+        patch_reconcile_started(&ctx, &latest_tenant).await;
     }
 
-    // Validate credential Secret if configured
-    // This only validates the Secret exists and has required keys.
-    // Actual credential injection happens via secretKeyRef in the StatefulSet.
-    if let Some(ref cfg) = latest_tenant.spec.creds_secret
-        && !cfg.name.is_empty()
-        && let Err(e) = ctx.validate_credential_secret(&latest_tenant).await
-    {
-        // Record event for credential validation failure
-        let _ = ctx
-            .record(
-                &latest_tenant,
-                EventType::Warning,
-                "CredentialValidationFailed",
-                &format!("Failed to validate credentials: {}", e),
-            )
-            .await;
-        return Err(e.into());
-    }
+    validate_tenant_prerequisites(&ctx, &latest_tenant).await?;
 
-    // Validate encryption / KMS: Vault requires endpoint + kmsSecret (and correct keys);
-    // must run whenever encryption is enabled — not only when kmsSecret is set, or Vault
-    // without a Secret reference would skip validation entirely.
-    if let Some(ref enc) = latest_tenant.spec.encryption
-        && enc.enabled
-        && let Err(e) = ctx.validate_kms_secret(&latest_tenant).await
-    {
-        let _ = ctx
-            .record(
-                &latest_tenant,
-                EventType::Warning,
-                "KmsSecretValidationFailed",
-                &format!("Failed to validate KMS secret: {}", e),
-            )
-            .await;
-        return Err(e.into());
-    }
+    maybe_cleanup_terminating_pods(&ctx, &latest_tenant, &ns).await?;
 
-    // Warn if Local backend has a kmsSecret configured (not used for Local)
-    if let Some(ref enc) = latest_tenant.spec.encryption
-        && enc.enabled
-        && enc.backend == crate::types::v1alpha1::encryption::KmsBackendType::Local
-        && enc.kms_secret.as_ref().is_some_and(|s| !s.name.is_empty())
-    {
-        let _ = ctx
-            .record(
-                &latest_tenant,
-                EventType::Warning,
-                "KmsConfigWarning",
-                "Local KMS backend does not use kmsSecret; the Secret reference will be ignored",
-            )
-            .await;
-    }
+    reconcile_rbac_resources(&ctx, &latest_tenant, &ns).await?;
 
-    // 0. Optional: unblock StatefulSet pods stuck terminating when their node is down.
-    // This is inspired by Longhorn's "Pod Deletion Policy When Node is Down".
-    if let Some(policy) = latest_tenant
-        .spec
-        .pod_deletion_policy_when_node_is_down
-        .clone()
-        && policy != crate::types::v1alpha1::k8s::PodDeletionPolicyWhenNodeIsDown::DoNothing
-    {
-        cleanup_stuck_terminating_pods_on_down_nodes(&latest_tenant, &ns, &ctx, policy).await?;
-    }
+    reconcile_services(&ctx, &latest_tenant, &ns).await?;
 
-    // 1. Create RBAC resources (conditionally based on service account settings)
-    let custom_sa = latest_tenant.spec.service_account_name.is_some();
-    let create_rbac = latest_tenant
-        .spec
-        .create_service_account_rbac
-        .unwrap_or(false);
-
-    if !custom_sa || create_rbac {
-        // Create Role
-        let role = ctx.apply(&latest_tenant.new_role(), &ns).await?;
-
-        if !custom_sa {
-            // Create default ServiceAccount and bind it
-            let service_account = ctx.apply(&latest_tenant.new_service_account(), &ns).await?;
-            ctx.apply(
-                &latest_tenant.new_role_binding(&service_account.name_any(), &role),
-                &ns,
-            )
-            .await?;
-        } else {
-            // Use custom ServiceAccount and bind it
-            let sa_name = latest_tenant.service_account_name();
-            ctx.apply(&latest_tenant.new_role_binding(&sa_name, &role), &ns)
-                .await?;
+    validate_no_pool_rename(&ctx, &latest_tenant, &ns).await?;
+
+    let summary = reconcile_pool_statefulsets(&ctx, &latest_tenant, &ns).await?;
+    finalize_tenant_status(&ctx, &latest_tenant, summary).await
+}
+
+#[cfg(test)]
+fn should_create_rbac(tenant: &Tenant) -> bool {
+    phases::should_create_rbac(tenant)
+}
+
+async fn context_result<T>(
+    result: Result<T, context::Error>,
+    ctx: &Context,
+    tenant: &Tenant,
+) -> Result<T, Error> {
+    match result {
+        Ok(value) => Ok(value),
+        Err(error) => {
+            let status_error = StatusError::from_context_error(&error);
+            patch_status_error(ctx, tenant, &status_error).await;
+            Err(error.into())
         }
     }
+}
 
-    // 2. Create Services
-    ctx.apply(&latest_tenant.new_io_service(), &ns).await?;
-    ctx.apply(&latest_tenant.new_console_service(), &ns).await?;
-    ctx.apply(&latest_tenant.new_headless_service(), &ns)
-        .await?;
-
-    // 3. Validate no pool renames (detect orphaned StatefulSets)
-    // Pool renames create new StatefulSets but leave old ones orphaned
-    let owned_statefulsets = ctx
-        .list::<k8s_openapi::api::apps::v1::StatefulSet>(&ns)
-        .await?;
-
-    let current_pool_names: std::collections::HashSet<_> = latest_tenant
-        .spec
-        .pools
-        .iter()
-        .map(|p| p.name.as_str())
-        .collect();
-
-    for ss in owned_statefulsets {
-        // Check if this StatefulSet is owned by this Tenant
-        if let Some(owner_refs) = &ss.metadata.owner_references {
-            let owned_by_tenant = owner_refs.iter().any(|owner| {
-                owner.kind == "Tenant"
-                    && owner.name == latest_tenant.name()
-                    && owner.uid == latest_tenant.metadata.uid.as_deref().unwrap_or("")
-            });
-
-            if owned_by_tenant {
-                let ss_name = ss.metadata.name.as_deref().unwrap_or("");
-                let tenant_prefix = format!("{}-", latest_tenant.name());
-
-                // Extract pool name from StatefulSet name (format: {tenant}-{pool})
-                if let Some(pool_name) = ss_name.strip_prefix(&tenant_prefix)
-                    && !current_pool_names.contains(pool_name)
-                {
-                    // Found orphaned StatefulSet - pool was renamed or removed
-                    return Err(types::error::Error::ImmutableFieldModified {
-                        name: latest_tenant.name(),
-                        field: "spec.pools[].name".to_string(),
-                        message: format!(
-                            "Pool name cannot be changed. Found StatefulSet '{}' for pool '{}' which no longer exists in spec. \
-                            Pool renames are not supported because they change the StatefulSet selector (immutable field). \
-                            To rename a pool: 1) Delete the Tenant, 2) Recreate with new pool names.",
-                            ss_name, pool_name
-                        ),
-                    }.into());
-                }
-            }
+async fn types_result<T>(
+    result: Result<T, types::error::Error>,
+    ctx: &Context,
+    tenant: &Tenant,
+) -> Result<T, Error> {
+    match result {
+        Ok(value) => Ok(value),
+        Err(error) => {
+            let status_error = StatusError::from_types_error(&error);
+            patch_status_error(ctx, tenant, &status_error).await;
+            Err(error.into())
         }
     }
+}
 
-    // 4. Create or update StatefulSets for each pool and collect their statuses
-    let mut pool_statuses = Vec::new();
-    let mut any_updating = false;
-    let mut any_degraded = false;
-    let mut total_replicas = 0;
-    let mut ready_replicas = 0;
-
-    for pool in &latest_tenant.spec.pools {
-        let ss_name = format!("{}-{}", latest_tenant.name(), pool.name);
-
-        // Try to get existing StatefulSet
-        match ctx
-            .get::<k8s_openapi::api::apps::v1::StatefulSet>(&ss_name, &ns)
-            .await
-        {
-            Ok(existing_ss) => {
-                // StatefulSet exists - check if update is needed
-                debug!("StatefulSet {} exists, checking if update needed", ss_name);
-
-                // First, validate that the update is safe (no immutable field changes)
-                if let Err(e) = latest_tenant.validate_statefulset_update(&existing_ss, pool) {
-                    error!("StatefulSet {} update validation failed: {}", ss_name, e);
-
-                    // Record event for validation failure
-                    let _ = ctx
-                        .record(
-                            &latest_tenant,
-                            EventType::Warning,
-                            "StatefulSetUpdateValidationFailed",
-                            &format!("Cannot update StatefulSet {}: {}", ss_name, e),
-                        )
-                        .await;
-
-                    return Err(e.into());
-                }
-
-                // Check if update is actually needed
-                if latest_tenant.statefulset_needs_update(&existing_ss, pool)? {
-                    debug!("StatefulSet {} needs update, applying changes", ss_name);
-
-                    // Record event for update start
-                    let _ = ctx
-                        .record(
-                            &latest_tenant,
-                            EventType::Normal,
-                            "StatefulSetUpdateStarted",
-                            &format!("Updating StatefulSet {}", ss_name),
-                        )
-                        .await;
-
-                    // Apply the update
-                    ctx.apply(&latest_tenant.new_statefulset(pool)?, &ns)
-                        .await?;
-
-                    debug!("StatefulSet {} updated successfully", ss_name);
-                } else {
-                    debug!("StatefulSet {} is up to date, no changes needed", ss_name);
-                }
-
-                // Fetch the StatefulSet again to get the latest status after any updates
-                let ss = ctx
-                    .get::<k8s_openapi::api::apps::v1::StatefulSet>(&ss_name, &ns)
-                    .await?;
-
-                // Build pool status from StatefulSet
-                let pool_status = latest_tenant.build_pool_status(&pool.name, &ss);
-
-                // Track if any pool is updating or degraded
-                use crate::types::v1alpha1::status::pool::PoolState;
-                match pool_status.state {
-                    PoolState::Updating => any_updating = true,
-                    PoolState::Degraded | PoolState::RolloutFailed => any_degraded = true,
-                    _ => {}
-                }
+async fn patch_status_error(ctx: &Context, tenant: &Tenant, status_error: &StatusError) {
+    let mut builder = StatusBuilder::from_tenant(tenant);
+    builder.mark_error(status_error);
+    let status = builder.build();
+    let should_record =
+        condition_marker_changed(tenant.status.as_ref(), &status, status_error.condition_type);
 
-                // Accumulate replica counts
-                if let Some(r) = pool_status.replicas {
-                    total_replicas += r;
-                }
-                if let Some(r) = pool_status.ready_replicas {
-                    ready_replicas += r;
-                }
-
-                pool_statuses.push(pool_status);
-            }
-            Err(e) if e.to_string().contains("NotFound") => {
-                // StatefulSet doesn't exist - create it
-                debug!("StatefulSet {} not found, creating", ss_name);
+    if should_record {
+        let _ = ctx
+            .record(
+                tenant,
+                status_error.event_type,
+                status_error.reason.as_str(),
+                &status_error.safe_message,
+            )
+            .await;
+    }
 
-                // Record event for creation
+    match ctx.patch_status_if_changed(tenant, status).await {
+        Ok(Some(_)) => {
+            info!(
+                tenant = %tenant.name(),
+                namespace = ?tenant.namespace(),
+                reason = status_error.reason.as_str(),
+                condition = status_error.condition_type.as_str(),
+                "patched Tenant status for reconcile error"
+            );
+        }
+        Ok(None) => {
+            debug!(
+                tenant = %tenant.name(),
+                namespace = ?tenant.namespace(),
+                reason = status_error.reason.as_str(),
+                "skipped Tenant status patch because error status is unchanged"
+            );
+        }
+        Err(error) => {
+            warn!(
+                tenant = %tenant.name(),
+                namespace = ?tenant.namespace(),
+                reason = status_error.reason.as_str(),
+                %error,
+                "failed to patch Tenant status for reconcile error"
+            );
+            if should_record {
                 let _ = ctx
                     .record(
-                        &latest_tenant,
-                        EventType::Normal,
-                        "StatefulSetCreated",
-                        &format!("Creating StatefulSet {}", ss_name),
+                        tenant,
+                        status_error.event_type,
+                        status_error.reason.as_str(),
+                        &status_error.safe_message,
                     )
                     .await;
-
-                ctx.apply(&latest_tenant.new_statefulset(pool)?, &ns)
-                    .await?;
-
-                debug!("StatefulSet {} created successfully", ss_name);
-
-                // After creation, fetch the StatefulSet to get its status
-                let ss = ctx
-                    .get::<k8s_openapi::api::apps::v1::StatefulSet>(&ss_name, &ns)
-                    .await?;
-                let pool_status = latest_tenant.build_pool_status(&pool.name, &ss);
-                any_updating = true; // New StatefulSet is always updating initially
-                pool_statuses.push(pool_status);
-            }
-            Err(e) => {
-                // Other error - propagate
-                error!("Failed to get StatefulSet {}: {}", ss_name, e);
-                return Err(e.into());
             }
+            let status_patch_error = StatusError::status_patch_failed(status_error.reason);
+            let _ = ctx
+                .record(
+                    tenant,
+                    status_patch_error.event_type,
+                    status_patch_error.reason.as_str(),
+                    &status_patch_error.safe_message,
+                )
+                .await;
         }
     }
+}
 
-    // 5. Aggregate pool statuses and determine overall Tenant conditions
-    use crate::types::v1alpha1::status::{Condition, Status};
-
-    let observed_generation = latest_tenant.metadata.generation;
-    let current_time = chrono::Utc::now().to_rfc3339_opts(chrono::SecondsFormat::Secs, true);
-
-    let mut conditions = Vec::new();
+async fn patch_reconcile_started(ctx: &Context, tenant: &Tenant) {
+    if !should_mark_reconcile_started(tenant) {
+        debug!(
+            tenant = %tenant.name(),
+            namespace = ?tenant.namespace(),
+            generation = ?tenant.metadata.generation,
+            observed_generation = ?tenant.status.as_ref().and_then(|status| status.observed_generation),
+            "skipping ReconcileStarted status patch because observed generation is current"
+        );
+        return;
+    }
 
-    // Determine Ready condition
-    let ready_condition = if any_degraded {
-        Condition {
-            type_: "Ready".to_string(),
-            status: "False".to_string(),
-            last_transition_time: Some(current_time.clone()),
-            observed_generation,
-            reason: "PoolDegraded".to_string(),
-            message: "One or more pools are degraded".to_string(),
+    let mut builder = StatusBuilder::from_tenant(tenant);
+    builder.mark_started();
+    let status = builder.build();
+
+    info!(
+        tenant = %tenant.name(),
+        namespace = ?tenant.namespace(),
+        generation = ?tenant.metadata.generation,
+        observed_generation = ?tenant.status.as_ref().and_then(|status| status.observed_generation),
+        "marking Tenant reconcile started for stale or missing status"
+    );
+
+    match ctx.patch_status_if_changed(tenant, status).await {
+        Ok(Some(_)) => {
+            info!(
+                tenant = %tenant.name(),
+                namespace = ?tenant.namespace(),
+                "patched Tenant ReconcileStarted status"
+            );
         }
-    } else if any_updating {
-        Condition {
-            type_: "Ready".to_string(),
-            status: "False".to_string(),
-            last_transition_time: Some(current_time.clone()),
-            observed_generation,
-            reason: "RolloutInProgress".to_string(),
-            message: "StatefulSet rollout in progress".to_string(),
+        Ok(None) => {
+            debug!(
+                tenant = %tenant.name(),
+                namespace = ?tenant.namespace(),
+                "ReconcileStarted status patch was a no-op"
+            );
         }
-    } else if ready_replicas == total_replicas && total_replicas > 0 {
-        Condition {
-            type_: "Ready".to_string(),
-            status: "True".to_string(),
-            last_transition_time: Some(current_time.clone()),
-            observed_generation,
-            reason: "AllPodsReady".to_string(),
-            message: format!("{}/{} pods ready", ready_replicas, total_replicas),
+        Err(error) => {
+            warn!(
+                tenant = %tenant.name(),
+                namespace = ?tenant.namespace(),
+                %error,
+                "failed to patch Tenant ReconcileStarted status"
+            );
+            let status_patch_error = StatusError::status_patch_failed(Reason::ReconcileStarted);
+            let _ = ctx
+                .record(
+                    tenant,
+                    status_patch_error.event_type,
+                    status_patch_error.reason.as_str(),
+                    &status_patch_error.safe_message,
+                )
+                .await;
         }
-    } else {
-        Condition {
-            type_: "Ready".to_string(),
-            status: "False".to_string(),
-            last_transition_time: Some(current_time.clone()),
-            observed_generation,
-            reason: "PodsNotReady".to_string(),
-            message: format!("{}/{} pods ready", ready_replicas, total_replicas),
-        }
-    };
-    conditions.push(ready_condition);
-
-    // Determine Progressing condition
-    if any_updating {
-        conditions.push(Condition {
-            type_: "Progressing".to_string(),
-            status: "True".to_string(),
-            last_transition_time: Some(current_time.clone()),
-            observed_generation,
-            reason: "RolloutInProgress".to_string(),
-            message: "StatefulSet rollout in progress".to_string(),
-        });
     }
+}
 
-    // Determine Degraded condition
-    if any_degraded {
-        conditions.push(Condition {
-            type_: "Degraded".to_string(),
-            status: "True".to_string(),
-            last_transition_time: Some(current_time.clone()),
-            observed_generation,
-            reason: "PoolDegraded".to_string(),
-            message: "One or more pools are degraded".to_string(),
-        });
+fn should_mark_reconcile_started(tenant: &Tenant) -> bool {
+    match (
+        tenant
+            .status
+            .as_ref()
+            .and_then(|status| status.observed_generation),
+        tenant.metadata.generation,
+    ) {
+        (Some(observed), Some(generation)) => observed < generation,
+        (None, Some(_)) => true,
+        (None, None) => tenant.status.is_none(),
+        (Some(_), None) => false,
     }
+}
 
-    // Determine overall state
-    let current_state = if any_degraded {
-        "Degraded".to_string()
-    } else if any_updating {
-        "Updating".to_string()
-    } else if ready_replicas == total_replicas && total_replicas > 0 {
-        "Ready".to_string()
-    } else {
-        "NotReady".to_string()
-    };
+async fn patch_status_and_record(
+    ctx: &Context,
+    tenant: &Tenant,
+    status: Status,
+    condition_type: ConditionType,
+    reason: Reason,
+    event_type: EventType,
+    message: &str,
+) -> Result<(), Error> {
+    let should_record = condition_marker_changed(tenant.status.as_ref(), &status, condition_type);
+    let patched = ctx.patch_status_if_changed(tenant, status).await?;
+    match patched {
+        Some(_) => {
+            info!(
+                tenant = %tenant.name(),
+                namespace = ?tenant.namespace(),
+                reason = reason.as_str(),
+                condition = condition_type.as_str(),
+                "patched Tenant status after reconciliation"
+            );
+            if should_record {
+                let _ = ctx
+                    .record(tenant, event_type, reason.as_str(), message)
+                    .await;
+            }
+        }
+        None => {
+            debug!(
+                tenant = %tenant.name(),
+                namespace = ?tenant.namespace(),
+                reason = reason.as_str(),
+                "skipped Tenant status patch because reconciled status is unchanged"
+            );
+        }
+    }
+    Ok(())
+}
 
-    // Build and update status
-    let status = Status {
-        current_state,
-        available_replicas: ready_replicas,
-        pools: pool_statuses,
-        observed_generation,
-        conditions,
-    };
+fn condition_marker_changed(
+    previous_status: Option<&Status>,
+    next_status: &Status,
+    condition_type: ConditionType,
+) -> bool {
+    condition_marker(previous_status, condition_type)
+        != condition_marker(Some(next_status), condition_type)
+}
 
-    debug!("Updating tenant status: {:?}", status);
-    ctx.update_status(&latest_tenant, status).await?;
+fn condition_marker(
+    status: Option<&Status>,
+    condition_type: ConditionType,
+) -> Option<(String, String)> {
+    status
+        .and_then(|status| status.condition(condition_type))
+        .map(|condition| (condition.status.clone(), condition.reason.clone()))
+}
 
-    // Requeue faster if any pool is updating
-    if any_updating {
-        debug!("Pools are updating, requeuing in 10 seconds");
-        Ok(Action::requeue(Duration::from_secs(10)))
-    } else {
-        Ok(Action::await_change())
-    }
+fn statefulset_owned_by_tenant(
+    ss: &k8s_openapi::api::apps::v1::StatefulSet,
+    tenant: &Tenant,
+) -> bool {
+    ss.metadata.owner_references.as_ref().is_some_and(|refs| {
+        refs.iter().any(|owner| {
+            owner.kind == "Tenant"
+                && owner.name == tenant.name()
+                && owner.uid == tenant.metadata.uid.as_deref().unwrap_or("")
+        })
+    })
 }
 
 async fn cleanup_stuck_terminating_pods_on_down_nodes(
@@ -626,7 +519,8 @@ pub fn error_policy(_object: Arc<Tenant>, error: &Error, _ctx: Arc<Context>) ->
             // Immutable field / invalid name errors - require user intervention
             // Use 60-second requeue to reduce event/log spam while user fixes the issue
             types::error::Error::ImmutableFieldModified { .. }
-            | types::error::Error::InvalidTenantName { .. } => {
+            | types::error::Error::InvalidTenantName { .. }
+            | types::error::Error::PoolDeleteBlocked { .. } => {
                 Action::requeue(Duration::from_secs(60))
             }
 
@@ -639,25 +533,49 @@ pub fn error_policy(_object: Arc<Tenant>, error: &Error, _ctx: Arc<Context>) ->
 #[cfg(test)]
 mod tests {
     use super::is_node_down;
-    use super::{pod_has_owner_kind, pod_matches_policy_controller_kind};
+    use super::{
+        pod_has_owner_kind, pod_matches_policy_controller_kind, should_create_rbac,
+        should_mark_reconcile_started,
+    };
+    use crate::types::v1alpha1::status::Status;
     use k8s_openapi::api::core::v1 as corev1;
     use k8s_openapi::apimachinery::pkg::apis::meta::v1 as metav1;
 
-    // Test 10: RBAC creation logic - default behavior
+    #[test]
+    fn should_not_mark_reconcile_started_when_generation_is_current() {
+        let mut tenant = crate::tests::create_test_tenant(None, None);
+        tenant.metadata.generation = Some(3);
+        tenant.status = Some(Status {
+            current_state: "Ready".to_string(),
+            observed_generation: Some(3),
+            ..Default::default()
+        });
+
+        assert!(!should_mark_reconcile_started(&tenant));
+    }
+
+    #[test]
+    fn should_mark_reconcile_started_for_missing_or_stale_status() {
+        let mut missing = crate::tests::create_test_tenant(None, None);
+        missing.metadata.generation = Some(3);
+        missing.status = None;
+        assert!(should_mark_reconcile_started(&missing));
+
+        let mut stale = crate::tests::create_test_tenant(None, None);
+        stale.metadata.generation = Some(3);
+        stale.status = Some(Status {
+            current_state: "Ready".to_string(),
+            observed_generation: Some(2),
+            ..Default::default()
+        });
+        assert!(should_mark_reconcile_started(&stale));
+    }
+
     #[test]
     fn test_should_create_rbac_default() {
         let tenant = crate::tests::create_test_tenant(None, None);
 
-        let custom_sa = tenant.spec.service_account_name.is_some();
-        let create_rbac = tenant.spec.create_service_account_rbac.unwrap_or(false);
-        let should_create_rbac = !custom_sa || create_rbac;
-
-        assert!(!custom_sa, "Should not have custom SA");
-        assert!(
-            !create_rbac,
-            "createServiceAccountRbac should default to false"
-        );
-        assert!(should_create_rbac, "Should create RBAC when no custom SA");
+        assert!(should_create_rbac(&tenant));
     }
 
     // Test 11: RBAC creation logic - custom SA with createServiceAccountRbac=true
@@ -665,16 +583,7 @@ mod tests {
     fn test_should_create_rbac_custom_sa_with_rbac() {
         let tenant = crate::tests::create_test_tenant(Some("my-custom-sa".to_string()), Some(true));
 
-        let custom_sa = tenant.spec.service_account_name.is_some();
-        let create_rbac = tenant.spec.create_service_account_rbac.unwrap_or(false);
-        let should_create_rbac = !custom_sa || create_rbac;
-
-        assert!(custom_sa, "Should have custom SA");
-        assert!(create_rbac, "createServiceAccountRbac should be true");
-        assert!(
-            should_create_rbac,
-            "Should create RBAC when explicitly requested"
-        );
+        assert!(should_create_rbac(&tenant));
     }
 
     // Test 12: RBAC creation logic - custom SA with createServiceAccountRbac=false
@@ -683,13 +592,7 @@ mod tests {
         let tenant =
             crate::tests::create_test_tenant(Some("my-custom-sa".to_string()), Some(false));
 
-        let custom_sa = tenant.spec.service_account_name.is_some();
-        let create_rbac = tenant.spec.create_service_account_rbac.unwrap_or(false);
-        let should_create_rbac = !custom_sa || create_rbac;
-
-        assert!(custom_sa, "Should have custom SA");
-        assert!(!create_rbac, "createServiceAccountRbac should be false");
-        assert!(!should_create_rbac, "Should skip RBAC creation");
+        assert!(!should_create_rbac(&tenant));
     }
 
     // Test 13: RBAC creation logic - custom SA with createServiceAccountRbac=None (default)
@@ -697,19 +600,7 @@ mod tests {
     fn test_should_skip_rbac_custom_sa_default() {
         let tenant = crate::tests::create_test_tenant(Some("my-custom-sa".to_string()), None);
 
-        let custom_sa = tenant.spec.service_account_name.is_some();
-        let create_rbac = tenant.spec.create_service_account_rbac.unwrap_or(false);
-        let should_create_rbac = !custom_sa || create_rbac;
-
-        assert!(custom_sa, "Should have custom SA");
-        assert!(
-            !create_rbac,
-            "createServiceAccountRbac should default to false"
-        );
-        assert!(
-            !should_create_rbac,
-            "Should skip RBAC when None treated as false"
-        );
+        assert!(!should_create_rbac(&tenant));
     }
 
     // Test 14: Service account determination in reconcile logic
diff --git a/src/reconcile/phases.rs b/src/reconcile/phases.rs
new file mode 100644
index 0000000..22f59d0
--- /dev/null
+++ b/src/reconcile/phases.rs
@@ -0,0 +1,529 @@
+// Copyright 2025 RustFS Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use super::{
+    Error, cleanup_stuck_terminating_pods_on_down_nodes, context, context_result,
+    patch_status_and_record, patch_status_error, statefulset_owned_by_tenant, types_result,
+};
+use crate::context::Context;
+use crate::status::{StatusBuilder, StatusError};
+use crate::types;
+use crate::types::v1alpha1::status::{ConditionType, Reason};
+use crate::types::v1alpha1::tenant::Tenant;
+use kube::ResourceExt;
+use kube::api::ListParams;
+use kube::runtime::controller::Action;
+use kube::runtime::events::EventType;
+use std::time::Duration;
+use tracing::{debug, error, warn};
+
+#[derive(Default)]
+pub(super) struct PoolReconcileSummary {
+    pool_statuses: Vec<crate::types::v1alpha1::status::pool::Pool>,
+    any_updating: bool,
+    any_degraded: bool,
+    total_replicas: i32,
+    ready_replicas: i32,
+}
+
+pub(super) async fn validate_tenant_prerequisites(
+    ctx: &Context,
+    tenant: &Tenant,
+) -> Result<(), Error> {
+    // Validate tenant name is DNS-1035 compliant (required for derived Service names).
+    if let Err(e) = tenant.validate_name() {
+        let status_error = StatusError::from_types_error(&e);
+        patch_status_error(ctx, tenant, &status_error).await;
+        return Err(e.into());
+    }
+
+    // Validate credential Secret if configured.
+    // This only validates the Secret exists and has required keys.
+    // Actual credential injection happens via secretKeyRef in the StatefulSet.
+    if let Some(ref cfg) = tenant.spec.creds_secret
+        && !cfg.name.is_empty()
+        && let Err(e) = ctx.validate_credential_secret(tenant).await
+    {
+        let status_error = StatusError::from_context_error(&e);
+        patch_status_error(ctx, tenant, &status_error).await;
+        return Err(e.into());
+    }
+
+    // Validate encryption / KMS: Vault requires endpoint + kmsSecret (and correct keys);
+    // must run whenever encryption is enabled — not only when kmsSecret is set, or Vault
+    // without a Secret reference would skip validation entirely.
+    if let Some(ref enc) = tenant.spec.encryption
+        && enc.enabled
+        && let Err(e) = ctx.validate_kms_secret(tenant).await
+    {
+        let status_error = StatusError::from_context_error(&e);
+        patch_status_error(ctx, tenant, &status_error).await;
+        return Err(e.into());
+    }
+
+    // Warn if Local backend has a kmsSecret configured (not used for Local).
+    if let Some(ref enc) = tenant.spec.encryption
+        && enc.enabled
+        && enc.backend == crate::types::v1alpha1::encryption::KmsBackendType::Local
+        && enc.kms_secret.as_ref().is_some_and(|s| !s.name.is_empty())
+    {
+        let _ = ctx
+            .record(
+                tenant,
+                EventType::Warning,
+                "KmsConfigWarning",
+                "Local KMS backend does not use kmsSecret; the Secret reference will be ignored",
+            )
+            .await;
+    }
+
+    Ok(())
+}
+
+pub(super) async fn maybe_cleanup_terminating_pods(
+    ctx: &Context,
+    tenant: &Tenant,
+    namespace: &str,
+) -> Result<(), Error> {
+    // Optional: unblock StatefulSet pods stuck terminating when their node is down.
+    // This is inspired by Longhorn's "Pod Deletion Policy When Node is Down".
+    if let Some(policy) = tenant.spec.pod_deletion_policy_when_node_is_down.clone()
+        && policy != crate::types::v1alpha1::k8s::PodDeletionPolicyWhenNodeIsDown::DoNothing
+    {
+        cleanup_stuck_terminating_pods_on_down_nodes(tenant, namespace, ctx, policy).await?;
+    }
+    Ok(())
+}
+
+pub(super) fn should_create_rbac(tenant: &Tenant) -> bool {
+    let custom_sa = tenant.spec.service_account_name.is_some();
+    let create_rbac = tenant.spec.create_service_account_rbac.unwrap_or(false);
+    !custom_sa || create_rbac
+}
+
+pub(super) async fn reconcile_rbac_resources(
+    ctx: &Context,
+    tenant: &Tenant,
+    namespace: &str,
+) -> Result<(), Error> {
+    if !should_create_rbac(tenant) {
+        return Ok(());
+    }
+
+    let role = context_result(ctx.apply(&tenant.new_role(), namespace).await, ctx, tenant).await?;
+
+    if tenant.spec.service_account_name.is_some() {
+        let sa_name = tenant.service_account_name();
+        context_result(
+            ctx.apply(&tenant.new_role_binding(&sa_name, &role), namespace)
+                .await,
+            ctx,
+            tenant,
+        )
+        .await?;
+    } else {
+        let service_account = context_result(
+            ctx.apply(&tenant.new_service_account(), namespace).await,
+            ctx,
+            tenant,
+        )
+        .await?;
+        context_result(
+            ctx.apply(
+                &tenant.new_role_binding(&service_account.name_any(), &role),
+                namespace,
+            )
+            .await,
+            ctx,
+            tenant,
+        )
+        .await?;
+    }
+
+    Ok(())
+}
+
+pub(super) async fn reconcile_services(
+    ctx: &Context,
+    tenant: &Tenant,
+    namespace: &str,
+) -> Result<(), Error> {
+    context_result(
+        ctx.apply(&tenant.new_io_service(), namespace).await,
+        ctx,
+        tenant,
+    )
+    .await?;
+    context_result(
+        ctx.apply(&tenant.new_console_service(), namespace).await,
+        ctx,
+        tenant,
+    )
+    .await?;
+    context_result(
+        ctx.apply(&tenant.new_headless_service(), namespace).await,
+        ctx,
+        tenant,
+    )
+    .await?;
+
+    Ok(())
+}
+
+pub(super) async fn validate_no_pool_rename(
+    ctx: &Context,
+    tenant: &Tenant,
+    namespace: &str,
+) -> Result<(), Error> {
+    let owned_statefulsets = context_result(
+        ctx.list_with_params::<k8s_openapi::api::apps::v1::StatefulSet>(
+            namespace,
+            &ListParams::default().labels(&format!("rustfs.tenant={}", tenant.name())),
+        )
+        .await,
+        ctx,
+        tenant,
+    )
+    .await?;
+
+    let current_pool_names: std::collections::HashSet<_> =
+        tenant.spec.pools.iter().map(|p| p.name.as_str()).collect();
+
+    let tenant_prefix = format!("{}-", tenant.name());
+    let existing_pool_names: std::collections::HashSet<String> = owned_statefulsets
+        .iter()
+        .filter(|ss| ss.metadata.deletion_timestamp.is_none())
+        .filter(|ss| statefulset_owned_by_tenant(ss, tenant))
+        .filter_map(|ss| {
+            ss.metadata
+                .name
+                .as_deref()
+                .and_then(|name| name.strip_prefix(&tenant_prefix))
+                .map(ToOwned::to_owned)
+        })
+        .collect();
+
+    let mut removed_pool_names: Vec<_> = existing_pool_names
+        .iter()
+        .filter(|pool_name| !current_pool_names.contains(pool_name.as_str()))
+        .cloned()
+        .collect();
+    removed_pool_names.sort_unstable();
+    let mut added_pool_names: Vec<_> = current_pool_names
+        .iter()
+        .filter(|pool_name| !existing_pool_names.contains::<str>(*pool_name))
+        .cloned()
+        .collect();
+    added_pool_names.sort_unstable();
+
+    if removed_pool_names.is_empty() {
+        return Ok(());
+    }
+
+    warn!(
+        tenant = %tenant.name(),
+        namespace = %namespace,
+        removed_pools = ?removed_pool_names,
+        added_pools = ?added_pool_names,
+        "detected pool removal or rename while owned StatefulSets still exist"
+    );
+    let err = if added_pool_names.is_empty() {
+        types::error::Error::PoolDeleteBlocked {
+            name: tenant.name(),
+            message: format!(
+                "Pool(s) '{}' were removed from spec while owned StatefulSets still exist. Restore the pool spec before starting a controlled decommission.",
+                removed_pool_names.join(",")
+            ),
+        }
+    } else {
+        types::error::Error::ImmutableFieldModified {
+            name: tenant.name(),
+            field: "spec.pools[].name".to_string(),
+            message: format!(
+                "Pool name cannot be changed. Removed pool(s) '{}' and added pool(s) '{}' in the same spec change.",
+                removed_pool_names.join(","),
+                added_pool_names.join(",")
+            ),
+        }
+    };
+    let status_error = StatusError::from_types_error(&err);
+    patch_status_error(ctx, tenant, &status_error).await;
+
+    Err(err.into())
+}
+
+pub(super) async fn reconcile_pool_statefulsets(
+    ctx: &Context,
+    tenant: &Tenant,
+    namespace: &str,
+) -> Result<PoolReconcileSummary, Error> {
+    let mut summary = PoolReconcileSummary::default();
+
+    for pool in &tenant.spec.pools {
+        let ss_name = format!("{}-{}", tenant.name(), pool.name);
+        match ctx
+            .get::<k8s_openapi::api::apps::v1::StatefulSet>(&ss_name, namespace)
+            .await
+        {
+            Ok(existing_ss) => {
+                reconcile_existing_pool_statefulset(
+                    ctx,
+                    tenant,
+                    namespace,
+                    pool,
+                    &ss_name,
+                    existing_ss,
+                    &mut summary,
+                )
+                .await?;
+            }
+            Err(e) if is_not_found_context_error(&e) => {
+                reconcile_missing_pool_statefulset(
+                    ctx,
+                    tenant,
+                    namespace,
+                    pool,
+                    &ss_name,
+                    &mut summary,
+                )
+                .await?;
+            }
+            Err(e) => {
+                error!("Failed to get StatefulSet {}: {}", ss_name, e);
+                let status_error = StatusError::from_context_error(&e);
+                patch_status_error(ctx, tenant, &status_error).await;
+                return Err(e.into());
+            }
+        }
+    }
+
+    Ok(summary)
+}
+
+fn is_not_found_context_error(error: &context::Error) -> bool {
+    matches!(
+        error,
+        context::Error::Kube {
+            source: kube::Error::Api(api_error)
+        } if api_error.code == 404
+    )
+}
+
+async fn reconcile_existing_pool_statefulset(
+    ctx: &Context,
+    tenant: &Tenant,
+    namespace: &str,
+    pool: &crate::types::v1alpha1::pool::Pool,
+    ss_name: &str,
+    existing_ss: k8s_openapi::api::apps::v1::StatefulSet,
+    summary: &mut PoolReconcileSummary,
+) -> Result<(), Error> {
+    debug!("StatefulSet {} exists, checking if update needed", ss_name);
+
+    if let Err(e) = tenant.validate_statefulset_update(&existing_ss, pool) {
+        error!("StatefulSet {} update validation failed: {}", ss_name, e);
+
+        let status_error = StatusError::statefulset_update_validation_failed(ss_name);
+        patch_status_error(ctx, tenant, &status_error).await;
+        return Err(e.into());
+    }
+
+    if types_result(
+        tenant.statefulset_needs_update(&existing_ss, pool),
+        ctx,
+        tenant,
+    )
+    .await?
+    {
+        debug!("StatefulSet {} needs update, applying changes", ss_name);
+
+        let _ = ctx
+            .record(
+                tenant,
+                EventType::Normal,
+                "StatefulSetUpdateStarted",
+                &format!("Updating StatefulSet {}", ss_name),
+            )
+            .await;
+
+        let desired = types_result(tenant.new_statefulset(pool), ctx, tenant).await?;
+        if let Err(e) = ctx.apply(&desired, namespace).await {
+            let status_error = StatusError::statefulset_apply_failed(ss_name);
+            patch_status_error(ctx, tenant, &status_error).await;
+            return Err(e.into());
+        }
+
+        debug!("StatefulSet {} updated successfully", ss_name);
+    } else {
+        debug!("StatefulSet {} is up to date, no changes needed", ss_name);
+    }
+
+    let ss = context_result(
+        ctx.get::<k8s_openapi::api::apps::v1::StatefulSet>(ss_name, namespace)
+            .await,
+        ctx,
+        tenant,
+    )
+    .await?;
+    let pool_status = tenant.build_pool_status(&pool.name, &ss);
+    update_pool_summary(summary, pool_status);
+
+    Ok(())
+}
+
+async fn reconcile_missing_pool_statefulset(
+    ctx: &Context,
+    tenant: &Tenant,
+    namespace: &str,
+    pool: &crate::types::v1alpha1::pool::Pool,
+    ss_name: &str,
+    summary: &mut PoolReconcileSummary,
+) -> Result<(), Error> {
+    debug!("StatefulSet {} not found, creating", ss_name);
+
+    let _ = ctx
+        .record(
+            tenant,
+            EventType::Normal,
+            "StatefulSetCreated",
+            &format!("Creating StatefulSet {}", ss_name),
+        )
+        .await;
+
+    let desired = types_result(tenant.new_statefulset(pool), ctx, tenant).await?;
+    if let Err(e) = ctx.apply(&desired, namespace).await {
+        let status_error = StatusError::statefulset_apply_failed(ss_name);
+        patch_status_error(ctx, tenant, &status_error).await;
+        return Err(e.into());
+    }
+
+    debug!("StatefulSet {} created successfully", ss_name);
+
+    let ss = context_result(
+        ctx.get::<k8s_openapi::api::apps::v1::StatefulSet>(ss_name, namespace)
+            .await,
+        ctx,
+        tenant,
+    )
+    .await?;
+    let pool_status = tenant.build_pool_status(&pool.name, &ss);
+    summary.any_updating = true; // New StatefulSet is always updating initially.
+    update_pool_summary(summary, pool_status);
+
+    Ok(())
+}
+
+fn update_pool_summary(
+    summary: &mut PoolReconcileSummary,
+    pool_status: crate::types::v1alpha1::status::pool::Pool,
+) {
+    use crate::types::v1alpha1::status::pool::PoolState;
+
+    match pool_status.state {
+        PoolState::Updating => summary.any_updating = true,
+        PoolState::Degraded | PoolState::RolloutFailed => summary.any_degraded = true,
+        _ => {}
+    }
+
+    if let Some(replicas) = pool_status.replicas {
+        summary.total_replicas += replicas;
+    }
+    if let Some(ready) = pool_status.ready_replicas {
+        summary.ready_replicas += ready;
+    }
+
+    summary.pool_statuses.push(pool_status);
+}
+
+pub(super) async fn finalize_tenant_status(
+    ctx: &Context,
+    tenant: &Tenant,
+    summary: PoolReconcileSummary,
+) -> Result<Action, Error> {
+    let mut builder = StatusBuilder::from_tenant(tenant);
+    builder.set_pool_statuses(summary.pool_statuses);
+
+    let (event_condition, event_reason, event_type, event_message) = if summary.any_degraded {
+        builder.finish_degraded(
+            Reason::PoolDegraded,
+            ConditionType::PoolsReady,
+            "One or more pools are degraded".to_string(),
+        );
+        (
+            ConditionType::PoolsReady,
+            Reason::PoolDegraded,
+            EventType::Warning,
+            "One or more pools are degraded".to_string(),
+        )
+    } else if summary.any_updating {
+        builder.finish_reconciling(
+            Reason::RolloutInProgress,
+            "StatefulSet rollout in progress".to_string(),
+        );
+        (
+            ConditionType::WorkloadsReady,
+            Reason::RolloutInProgress,
+            EventType::Normal,
+            "StatefulSet rollout in progress".to_string(),
+        )
+    } else if summary.ready_replicas == summary.total_replicas && summary.total_replicas > 0 {
+        builder.finish_success();
+        (
+            ConditionType::Ready,
+            Reason::ReconcileSucceeded,
+            EventType::Normal,
+            format!(
+                "{}/{} pods ready",
+                summary.ready_replicas, summary.total_replicas
+            ),
+        )
+    } else {
+        builder.finish_reconciling(
+            Reason::PodsNotReady,
+            format!(
+                "{}/{} pods ready",
+                summary.ready_replicas, summary.total_replicas
+            ),
+        );
+        (
+            ConditionType::WorkloadsReady,
+            Reason::PodsNotReady,
+            EventType::Normal,
+            format!(
+                "{}/{} pods ready",
+                summary.ready_replicas, summary.total_replicas
+            ),
+        )
+    };
+
+    let status = builder.build();
+    debug!("Patching tenant status if changed: {:?}", status);
+    patch_status_and_record(
+        ctx,
+        tenant,
+        status,
+        event_condition,
+        event_reason,
+        event_type,
+        &event_message,
+    )
+    .await?;
+
+    if summary.any_updating {
+        debug!("Pools are updating, requeuing in 10 seconds");
+        Ok(Action::requeue(Duration::from_secs(10)))
+    } else {
+        Ok(Action::await_change())
+    }
+}
diff --git a/src/status.rs b/src/status.rs
new file mode 100644
index 0000000..5f1ac2d
--- /dev/null
+++ b/src/status.rs
@@ -0,0 +1,888 @@
+// Copyright 2025 RustFS Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use crate::context;
+use crate::types;
+use crate::types::v1alpha1::status::{
+    ConditionInput, ConditionStatus, ConditionType, Reason, Status, is_blocked_reason, pool,
+    summarize_current_state,
+};
+use crate::types::v1alpha1::tenant::Tenant;
+use kube::runtime::events::EventType;
+
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+pub enum StatusImpact {
+    UserBlocked,
+    Degraded,
+    Transient,
+}
+
+#[derive(Clone, Debug)]
+pub struct StatusError {
+    pub reason: Reason,
+    pub condition_type: ConditionType,
+    pub impact: StatusImpact,
+    pub safe_message: String,
+    pub event_type: EventType,
+}
+
+impl StatusError {
+    pub fn from_context_error(error: &context::Error) -> Self {
+        match error {
+            context::Error::CredentialSecretNotFound { name } => Self::blocked(
+                Reason::CredentialSecretNotFound,
+                ConditionType::CredentialsReady,
+                format!("Credential Secret '{}' was not found", name),
+            ),
+            context::Error::CredentialSecretMissingKey { secret_name, key } => Self::blocked(
+                Reason::CredentialSecretMissingKey,
+                ConditionType::CredentialsReady,
+                format!(
+                    "Credential Secret '{}' is missing required key '{}'",
+                    secret_name, key
+                ),
+            ),
+            context::Error::CredentialSecretInvalidEncoding { secret_name, key } => Self::blocked(
+                Reason::CredentialSecretInvalidEncoding,
+                ConditionType::CredentialsReady,
+                format!(
+                    "Credential Secret '{}' key '{}' must contain valid UTF-8",
+                    secret_name, key
+                ),
+            ),
+            context::Error::CredentialSecretTooShort {
+                secret_name, key, ..
+            } => Self::blocked(
+                Reason::CredentialSecretTooShort,
+                ConditionType::CredentialsReady,
+                format!(
+                    "Credential Secret '{}' key '{}' must be at least 8 characters",
+                    secret_name, key
+                ),
+            ),
+            context::Error::KmsSecretNotFound { name } => Self::blocked(
+                Reason::KmsSecretNotFound,
+                ConditionType::KmsReady,
+                format!("KMS Secret '{}' was not found", name),
+            ),
+            context::Error::KmsSecretMissingKey { secret_name, key } => Self::blocked(
+                Reason::KmsSecretMissingKey,
+                ConditionType::KmsReady,
+                format!(
+                    "KMS Secret '{}' is missing required key '{}'",
+                    secret_name, key
+                ),
+            ),
+            context::Error::KmsConfigInvalid { message } => Self::blocked(
+                Reason::KmsConfigInvalid,
+                ConditionType::KmsReady,
+                sanitize_message(message),
+            ),
+            context::Error::Types { source } => Self::from_types_error(source),
+            context::Error::Kube { .. } => Self::transient(
+                Reason::KubernetesApiError,
+                ConditionType::Ready,
+                "Kubernetes API request failed".to_string(),
+            ),
+            context::Error::Record { .. } => Self::transient(
+                Reason::KubernetesApiError,
+                ConditionType::Ready,
+                "Kubernetes Event recording failed".to_string(),
+            ),
+            context::Error::Serde { .. } => Self::degraded(
+                Reason::KubernetesApiError,
+                ConditionType::Ready,
+                "Failed to serialize Kubernetes status patch".to_string(),
+            ),
+        }
+    }
+
+    pub fn status_patch_failed(reason: Reason) -> Self {
+        Self {
+            reason: Reason::StatusPatchFailed,
+            condition_type: ConditionType::Ready,
+            impact: StatusImpact::Transient,
+            safe_message: format!(
+                "Failed to patch Tenant status for reason {}",
+                reason.as_str()
+            ),
+            event_type: EventType::Warning,
+        }
+    }
+
+    pub fn from_types_error(error: &types::error::Error) -> Self {
+        match error {
+            types::error::Error::InvalidTenantName { reason, .. } => Self::blocked(
+                Reason::InvalidTenantName,
+                ConditionType::SpecValid,
+                sanitize_message(reason),
+            ),
+            types::error::Error::ImmutableFieldModified { field, .. } => Self::blocked(
+                Reason::ImmutableFieldModified,
+                ConditionType::SpecValid,
+                format!("Immutable field '{}' was modified", field),
+            ),
+            types::error::Error::PoolDeleteBlocked { message, .. } => Self::blocked(
+                Reason::PoolDeleteBlocked,
+                ConditionType::SpecValid,
+                sanitize_message(message),
+            ),
+            types::error::Error::NoNamespace => Self::transient(
+                Reason::KubernetesApiError,
+                ConditionType::Ready,
+                "Tenant namespace is not available".to_string(),
+            ),
+            types::error::Error::InternalError { msg } => Self::degraded(
+                Reason::KubernetesApiError,
+                ConditionType::Ready,
+                sanitize_message(msg),
+            ),
+            types::error::Error::SerdeJson { .. } => Self::degraded(
+                Reason::KubernetesApiError,
+                ConditionType::Ready,
+                "Failed to serialize Kubernetes object".to_string(),
+            ),
+        }
+    }
+
+    pub fn statefulset_apply_failed(name: &str) -> Self {
+        Self::degraded(
+            Reason::StatefulSetApplyFailed,
+            ConditionType::WorkloadsReady,
+            format!("Failed to apply StatefulSet '{}'", name),
+        )
+    }
+
+    pub fn statefulset_update_validation_failed(name: &str) -> Self {
+        Self::blocked(
+            Reason::StatefulSetUpdateValidationFailed,
+            ConditionType::WorkloadsReady,
+            format!("StatefulSet '{}' update validation failed", name),
+        )
+    }
+
+    fn blocked(reason: Reason, condition_type: ConditionType, safe_message: String) -> Self {
+        Self {
+            reason,
+            condition_type,
+            impact: StatusImpact::UserBlocked,
+            safe_message,
+            event_type: EventType::Warning,
+        }
+    }
+
+    fn degraded(reason: Reason, condition_type: ConditionType, safe_message: String) -> Self {
+        Self {
+            reason,
+            condition_type,
+            impact: StatusImpact::Degraded,
+            safe_message,
+            event_type: EventType::Warning,
+        }
+    }
+
+    fn transient(reason: Reason, condition_type: ConditionType, safe_message: String) -> Self {
+        Self {
+            reason,
+            condition_type,
+            impact: StatusImpact::Transient,
+            safe_message,
+            event_type: EventType::Warning,
+        }
+    }
+}
+
+pub struct StatusBuilder {
+    generation: Option<i64>,
+    now: String,
+    next: Status,
+}
+
+impl StatusBuilder {
+    pub fn from_tenant(tenant: &Tenant) -> Self {
+        Self {
+            generation: tenant.metadata.generation,
+            now: chrono::Utc::now().to_rfc3339_opts(chrono::SecondsFormat::Secs, true),
+            next: tenant.status.clone().unwrap_or_default(),
+        }
+    }
+
+    pub fn set_pool_statuses(&mut self, pools: Vec<pool::Pool>) {
+        self.next.available_replicas = pools.iter().filter_map(|pool| pool.ready_replicas).sum();
+        self.next.pools = pools;
+    }
+
+    pub fn mark_started(&mut self) {
+        self.set_condition(
+            ConditionType::Ready,
+            ConditionStatus::False,
+            Reason::ReconcileStarted,
+            "Reconcile has started".to_string(),
+        );
+        self.set_condition(
+            ConditionType::Reconciling,
+            ConditionStatus::True,
+            Reason::ReconcileStarted,
+            "Operator is reconciling the latest Tenant generation".to_string(),
+        );
+        self.set_condition(
+            ConditionType::Degraded,
+            ConditionStatus::False,
+            Reason::ReconcileStarted,
+            "No persistent degradation has been confirmed".to_string(),
+        );
+    }
+
+    pub fn mark_error(&mut self, error: &StatusError) {
+        self.clear_stale_blocked_conditions(
+            error.condition_type,
+            error.reason,
+            &error.safe_message,
+        );
+
+        match error.impact {
+            StatusImpact::UserBlocked => {
+                self.set_condition(
+                    ConditionType::Ready,
+                    ConditionStatus::False,
+                    error.reason,
+                    error.safe_message.clone(),
+                );
+                self.set_condition(
+                    ConditionType::Reconciling,
+                    ConditionStatus::False,
+                    error.reason,
+                    "Reconcile is blocked by user-fixable configuration".to_string(),
+                );
+                self.set_condition(
+                    ConditionType::Degraded,
+                    ConditionStatus::True,
+                    error.reason,
+                    error.safe_message.clone(),
+                );
+                self.set_condition(
+                    error.condition_type,
+                    ConditionStatus::False,
+                    error.reason,
+                    error.safe_message.clone(),
+                );
+            }
+            StatusImpact::Degraded => {
+                self.finish_degraded(
+                    error.reason,
+                    error.condition_type,
+                    error.safe_message.clone(),
+                );
+                self.set_condition(
+                    error.condition_type,
+                    ConditionStatus::False,
+                    error.reason,
+                    error.safe_message.clone(),
+                );
+            }
+            StatusImpact::Transient => {
+                self.set_condition(
+                    ConditionType::Ready,
+                    ConditionStatus::Unknown,
+                    error.reason,
+                    error.safe_message.clone(),
+                );
+                self.set_condition(
+                    ConditionType::Reconciling,
+                    ConditionStatus::True,
+                    error.reason,
+                    "Reconcile will retry after a Kubernetes API error".to_string(),
+                );
+                self.set_condition(
+                    ConditionType::Degraded,
+                    ConditionStatus::False,
+                    error.reason,
+                    "No persistent degradation has been confirmed".to_string(),
+                );
+                self.set_condition(
+                    error.condition_type,
+                    ConditionStatus::Unknown,
+                    error.reason,
+                    error.safe_message.clone(),
+                );
+            }
+        }
+    }
+
+    pub fn finish_success(&mut self) {
+        self.mark_default_components_ready();
+        self.set_condition(
+            ConditionType::Ready,
+            ConditionStatus::True,
+            Reason::ReconcileSucceeded,
+            "Tenant is ready".to_string(),
+        );
+        self.set_condition(
+            ConditionType::Reconciling,
+            ConditionStatus::False,
+            Reason::ReconcileSucceeded,
+            "Reconcile completed successfully".to_string(),
+        );
+        self.set_condition(
+            ConditionType::Degraded,
+            ConditionStatus::False,
+            Reason::ReconcileSucceeded,
+            "Tenant is not degraded".to_string(),
+        );
+    }
+
+    pub fn finish_reconciling(&mut self, reason: Reason, message: String) {
+        self.mark_default_components_ready();
+        self.set_condition(
+            ConditionType::Ready,
+            ConditionStatus::False,
+            reason,
+            message.clone(),
+        );
+        self.set_condition(
+            ConditionType::Reconciling,
+            ConditionStatus::True,
+            reason,
+            message.clone(),
+        );
+        self.set_condition(
+            ConditionType::Degraded,
+            ConditionStatus::False,
+            reason,
+            "Tenant is progressing".to_string(),
+        );
+        self.set_condition(
+            ConditionType::WorkloadsReady,
+            ConditionStatus::False,
+            reason,
+            message,
+        );
+    }
+
+    pub fn finish_degraded(
+        &mut self,
+        reason: Reason,
+        condition_type: ConditionType,
+        message: String,
+    ) {
+        self.mark_default_components_ready();
+        self.set_condition(
+            ConditionType::Ready,
+            ConditionStatus::False,
+            reason,
+            message.clone(),
+        );
+        self.set_condition(
+            ConditionType::Reconciling,
+            ConditionStatus::False,
+            reason,
+            "Reconcile is not actively progressing".to_string(),
+        );
+        self.set_condition(
+            ConditionType::Degraded,
+            ConditionStatus::True,
+            reason,
+            message.clone(),
+        );
+        self.set_condition(condition_type, ConditionStatus::False, reason, message);
+    }
+
+    pub fn build(mut self) -> Status {
+        self.next.observed_generation = self.generation;
+        self.next.current_state = summarize_current_state(&self.next);
+        self.next.sort_conditions();
+        self.next
+    }
+
+    fn mark_default_components_ready(&mut self) {
+        for condition_type in [
+            ConditionType::SpecValid,
+            ConditionType::CredentialsReady,
+            ConditionType::KmsReady,
+            ConditionType::PoolsReady,
+            ConditionType::WorkloadsReady,
+        ] {
+            self.set_condition(
+                condition_type,
+                ConditionStatus::True,
+                Reason::ReconcileSucceeded,
+                format!("{} is ready", condition_type.as_str()),
+            );
+        }
+    }
+
+    fn clear_stale_blocked_conditions(
+        &mut self,
+        current_condition_type: ConditionType,
+        reason: Reason,
+        message: &str,
+    ) {
+        let current_type = current_condition_type.as_str();
+        for condition in &mut self.next.conditions {
+            if condition.type_ == current_type
+                || condition.status != ConditionStatus::False.as_str()
+                || !is_blocked_reason(&condition.reason)
+            {
+                continue;
+            }
+
+            if condition.status != ConditionStatus::Unknown.as_str() {
+                condition.last_transition_time = Some(self.now.clone());
+            }
+            condition.status = ConditionStatus::Unknown.as_str().to_string();
+            condition.reason = reason.as_str().to_string();
+            condition.message = format!(
+                "Condition was not confirmed during the current reconcile: {}",
+                message
+            );
+            condition.observed_generation = self.generation;
+        }
+    }
+
+    fn set_condition(
+        &mut self,
+        type_: ConditionType,
+        status: ConditionStatus,
+        reason: Reason,
+        message: String,
+    ) {
+        self.next.upsert_condition(ConditionInput {
+            type_,
+            status,
+            reason,
+            message,
+            observed_generation: self.generation,
+            now: self.now.clone(),
+        });
+    }
+}
+
+fn sanitize_message(message: &str) -> String {
+    redact_sensitive_pairs(message)
+}
+
+fn redact_sensitive_pairs(message: &str) -> String {
+    const SENSITIVE_KEYS: [&str; 4] = ["token", "password", "accesskey", "secretkey"];
+
+    fn is_sensitive_key(key: &str) -> bool {
+        matches!(key, "token" | "password" | "accesskey" | "secretkey")
+    }
+
+    fn normalize_key(raw: &str) -> String {
+        raw.trim_matches(|ch: char| !ch.is_ascii_alphanumeric() && ch != '-' && ch != '_')
+            .to_ascii_lowercase()
+    }
+
+    fn parse_value_end(input: &str, start: usize) -> usize {
+        let bytes = input.as_bytes();
+        if start >= bytes.len() {
+            return start;
+        }
+        let first = bytes[start];
+        if first == b'"' || first == b'\'' {
+            let quote = first;
+            let mut index = start + 1;
+            while index < bytes.len() {
+                if bytes[index] == quote && bytes.get(index.wrapping_sub(1)) != Some(&b'\\') {
+                    return index + 1;
+                }
+                index += 1;
+            }
+            return bytes.len();
+        }
+
+        let mut index = start;
+        while index < bytes.len() {
+            let ch = bytes[index] as char;
+            if ch.is_whitespace() || matches!(ch, ',' | ';' | '}' | ']' | ')') {
+                break;
+            }
+            index += 1;
+        }
+        index
+    }
+
+    fn redacted_value(original: &str) -> String {
+        if original.len() >= 2 {
+            let bytes = original.as_bytes();
+            let first = bytes[0];
+            let last = bytes[bytes.len() - 1];
+            if (first == b'"' && last == b'"') || (first == b'\'' && last == b'\'') {
+                let quote = first as char;
+                return format!("{quote}<redacted>{quote}");
+            }
+        }
+        "<redacted>".to_string()
+    }
+
+    let bytes = message.as_bytes();
+    let mut output = String::with_capacity(message.len());
+    let mut cursor = 0usize;
+
+    while cursor < bytes.len() {
+        let mut matched = false;
+
+        for key in SENSITIVE_KEYS {
+            let key_len = key.len();
+
+            let unquoted_match = cursor + key_len <= bytes.len()
+                && message[cursor..cursor + key_len].eq_ignore_ascii_case(key);
+            let quoted_match = cursor + key_len + 2 <= bytes.len()
+                && matches!(bytes[cursor] as char, '"' | '\'')
+                && bytes[cursor + key_len + 1] == bytes[cursor]
+                && message[cursor + 1..cursor + 1 + key_len].eq_ignore_ascii_case(key);
+
+            let (key_start, key_end, cursor_after_key) = if unquoted_match {
+                if cursor > 0 {
+                    let prev = bytes[cursor - 1] as char;
+                    if prev.is_ascii_alphanumeric() || prev == '_' || prev == '-' {
+                        continue;
+                    }
+                }
+                (cursor, cursor + key_len, cursor + key_len)
+            } else if quoted_match {
+                let key_start = cursor + 1;
+                (key_start, key_start + key_len, key_start + key_len + 1)
+            } else {
+                continue;
+            };
+
+            let candidate = &message[key_start..key_end];
+
+            let mut sep_index = cursor_after_key;
+            while sep_index < bytes.len() && (bytes[sep_index] as char).is_whitespace() {
+                sep_index += 1;
+            }
+            if sep_index >= bytes.len() || !matches!(bytes[sep_index] as char, '=' | ':') {
+                continue;
+            }
+
+            let mut value_start = sep_index + 1;
+            while value_start < bytes.len() && (bytes[value_start] as char).is_whitespace() {
+                value_start += 1;
+            }
+            let value_end = parse_value_end(message, value_start);
+            if value_end <= value_start {
+                continue;
+            }
+
+            let normalized = normalize_key(candidate);
+            if !is_sensitive_key(&normalized) {
+                continue;
+            }
+
+            output.push_str(&message[cursor..value_start]);
+            output.push_str(&redacted_value(&message[value_start..value_end]));
+            cursor = value_end;
+            matched = true;
+            break;
+        }
+
+        if !matched {
+            output.push(bytes[cursor] as char);
+            cursor += 1;
+        }
+    }
+
+    output
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::types::v1alpha1::status::Condition;
+
+    #[test]
+    fn status_builder_maps_credential_missing_key() {
+        let tenant = crate::tests::create_test_tenant(None, None);
+        let err = context::Error::CredentialSecretMissingKey {
+            secret_name: "creds".to_string(),
+            key: "accesskey".to_string(),
+        };
+
+        let status_error = StatusError::from_context_error(&err);
+        let mut builder = StatusBuilder::from_tenant(&tenant);
+        builder.mark_error(&status_error);
+        let status = builder.build();
+
+        let condition = status.condition(ConditionType::CredentialsReady).unwrap();
+        assert_eq!(condition.status, "False");
+        assert_eq!(condition.reason, "CredentialSecretMissingKey");
+        assert_eq!(status.current_state, "Blocked");
+        assert!(status.condition(ConditionType::KmsReady).is_none());
+        assert!(status.condition(ConditionType::WorkloadsReady).is_none());
+    }
+
+    #[test]
+    fn transition_time_is_preserved_when_status_does_not_change() {
+        let mut status = Status {
+            conditions: vec![Condition {
+                type_: "CredentialsReady".to_string(),
+                status: "False".to_string(),
+                last_transition_time: Some("old".to_string()),
+                observed_generation: Some(1),
+                reason: "CredentialSecretNotFound".to_string(),
+                message: "old".to_string(),
+            }],
+            ..Default::default()
+        };
+
+        status.upsert_condition(ConditionInput {
+            type_: ConditionType::CredentialsReady,
+            status: ConditionStatus::False,
+            reason: Reason::CredentialSecretMissingKey,
+            message: "new".to_string(),
+            observed_generation: Some(2),
+            now: "new-time".to_string(),
+        });
+
+        let condition = status.condition(ConditionType::CredentialsReady).unwrap();
+        assert_eq!(condition.last_transition_time.as_deref(), Some("old"));
+        assert_eq!(condition.reason, "CredentialSecretMissingKey");
+        assert_eq!(condition.observed_generation, Some(2));
+    }
+
+    #[test]
+    fn conditions_are_sorted_by_core_priority_then_name() {
+        let mut status = Status {
+            conditions: vec![
+                condition("ZFeatureReady", "True", "ReconcileSucceeded"),
+                condition("CredentialsReady", "True", "ReconcileSucceeded"),
+                condition("Ready", "True", "ReconcileSucceeded"),
+                condition("AFeatureReady", "True", "ReconcileSucceeded"),
+                condition("Degraded", "False", "ReconcileSucceeded"),
+            ],
+            ..Default::default()
+        };
+
+        status.sort_conditions();
+        let types: Vec<_> = status
+            .conditions
+            .iter()
+            .map(|condition| condition.type_.as_str())
+            .collect();
+
+        assert_eq!(
+            types,
+            vec![
+                "Ready",
+                "Degraded",
+                "CredentialsReady",
+                "AFeatureReady",
+                "ZFeatureReady"
+            ]
+        );
+    }
+
+    #[test]
+    fn blocked_reason_wins_current_state_summary() {
+        let status = Status {
+            conditions: vec![
+                condition("Reconciling", "True", "RolloutInProgress"),
+                condition("CredentialsReady", "False", "CredentialSecretNotFound"),
+                condition("Degraded", "True", "CredentialSecretNotFound"),
+            ],
+            ..Default::default()
+        };
+
+        assert_eq!(
+            crate::types::v1alpha1::status::summarize_current_state(&status),
+            "Blocked"
+        );
+        assert_eq!(
+            crate::types::v1alpha1::status::primary_condition(&status)
+                .map(|condition| condition.reason.as_str()),
+            Some("CredentialSecretNotFound")
+        );
+    }
+
+    #[test]
+    fn transient_error_does_not_keep_old_blocked_component_primary() {
+        let mut tenant = crate::tests::create_test_tenant(None, None);
+        tenant.metadata.generation = Some(7);
+        tenant.status = Some(Status {
+            current_state: "Blocked".to_string(),
+            available_replicas: 0,
+            pools: Vec::new(),
+            observed_generation: Some(7),
+            conditions: vec![condition(
+                "CredentialsReady",
+                "False",
+                "CredentialSecretNotFound",
+            )],
+        });
+        let status_error = StatusError {
+            reason: Reason::KubernetesApiError,
+            condition_type: ConditionType::Ready,
+            impact: StatusImpact::Transient,
+            safe_message: "Kubernetes API request failed".to_string(),
+            event_type: EventType::Warning,
+        };
+
+        let mut builder = StatusBuilder::from_tenant(&tenant);
+        builder.mark_error(&status_error);
+        let status = builder.build();
+
+        assert_eq!(status.current_state, "Reconciling");
+        assert_eq!(
+            crate::types::v1alpha1::status::primary_condition(&status)
+                .map(|condition| condition.reason.as_str()),
+            Some("KubernetesApiError")
+        );
+        assert_eq!(
+            status
+                .condition(ConditionType::CredentialsReady)
+                .map(|condition| condition.status.as_str()),
+            Some("Unknown")
+        );
+    }
+
+    #[test]
+    fn next_actions_are_registry_driven() {
+        assert_eq!(
+            crate::types::v1alpha1::status::next_actions_for_reason("PoolDeleteBlocked"),
+            vec!["restorePoolSpec", "startDecommissionAfterRestore"]
+        );
+        assert!(
+            crate::types::v1alpha1::status::next_actions_for_reason("UnknownReason").is_empty()
+        );
+    }
+
+    #[test]
+    fn degraded_status_targets_requested_component() {
+        let tenant = crate::tests::create_test_tenant(None, None);
+        let mut builder = StatusBuilder::from_tenant(&tenant);
+        builder.finish_degraded(
+            Reason::StatefulSetApplyFailed,
+            ConditionType::WorkloadsReady,
+            "failed".to_string(),
+        );
+        let status = builder.build();
+
+        assert_eq!(
+            status
+                .condition(ConditionType::WorkloadsReady)
+                .map(|condition| condition.status.as_str()),
+            Some("False")
+        );
+        assert_eq!(
+            status
+                .condition(ConditionType::PoolsReady)
+                .map(|condition| condition.status.as_str()),
+            Some("True")
+        );
+    }
+
+    #[test]
+    fn mark_started_sets_reconciling_condition() {
+        let tenant = crate::tests::create_test_tenant(None, None);
+        let mut builder = StatusBuilder::from_tenant(&tenant);
+        builder.mark_started();
+        let status = builder.build();
+
+        assert_eq!(status.current_state, "Reconciling");
+        assert_eq!(
+            status
+                .condition(ConditionType::Reconciling)
+                .map(|condition| condition.reason.as_str()),
+            Some("ReconcileStarted")
+        );
+    }
+
+    #[test]
+    fn transient_error_does_not_keep_previous_blocked_condition_current() {
+        let mut tenant = crate::tests::create_test_tenant(None, None);
+        tenant.metadata.generation = Some(7);
+        tenant.status = Some(Status {
+            current_state: "Blocked".to_string(),
+            observed_generation: Some(7),
+            conditions: vec![
+                condition("CredentialsReady", "False", "CredentialSecretNotFound"),
+                condition("Degraded", "True", "CredentialSecretNotFound"),
+            ],
+            ..Default::default()
+        });
+        let status_error = StatusError {
+            reason: Reason::KubernetesApiError,
+            condition_type: ConditionType::Ready,
+            impact: StatusImpact::Transient,
+            safe_message: "Kubernetes API request failed".to_string(),
+            event_type: EventType::Warning,
+        };
+
+        let mut builder = StatusBuilder::from_tenant(&tenant);
+        builder.mark_error(&status_error);
+        let status = builder.build();
+
+        assert_eq!(status.current_state, "Reconciling");
+        assert_eq!(
+            crate::types::v1alpha1::status::primary_condition(&status)
+                .map(|condition| condition.reason.as_str()),
+            Some("KubernetesApiError")
+        );
+        assert_eq!(
+            status
+                .condition(ConditionType::CredentialsReady)
+                .map(|condition| condition.status.as_str()),
+            Some("Unknown")
+        );
+    }
+
+    #[test]
+    fn sanitize_message_preserves_required_key_names() {
+        let message = "Vault backend requires kmsSecret referencing a Secret with key vault-token";
+
+        assert_eq!(sanitize_message(message), message);
+    }
+
+    #[test]
+    fn sanitize_message_redacts_colon_and_json_secret_values() {
+        let message =
+            "kms config token: tok_123 password: p@ss accesskey: AKIA_TEST secretkey: SK_TEST";
+
+        let sanitized = sanitize_message(message);
+
+        assert!(sanitized.contains("token"));
+        assert!(sanitized.contains("password"));
+        assert!(sanitized.contains("accesskey"));
+        assert!(sanitized.contains("secretkey"));
+        assert!(!sanitized.contains("tok_123"));
+        assert!(!sanitized.contains("p@ss"));
+        assert!(!sanitized.contains("AKIA_TEST"));
+        assert!(!sanitized.contains("SK_TEST"));
+    }
+
+    #[test]
+    fn sanitize_message_redacts_json_key_value_pairs() {
+        let message = "{\"accesskey\":\"AKIA_JSON\",\"secretkey\":\"SECRET_JSON\"}";
+
+        let sanitized = sanitize_message(message);
+
+        assert!(sanitized.contains("\"accesskey\""));
+        assert!(sanitized.contains("\"secretkey\""));
+        assert!(!sanitized.contains("AKIA_JSON"));
+        assert!(!sanitized.contains("SECRET_JSON"));
+    }
+
+    fn condition(type_: &str, status: &str, reason: &str) -> Condition {
+        Condition {
+            type_: type_.to_string(),
+            status: status.to_string(),
+            last_transition_time: Some("now".to_string()),
+            observed_generation: Some(1),
+            reason: reason.to_string(),
+            message: reason.to_string(),
+        }
+    }
+}
diff --git a/src/types/error.rs b/src/types/error.rs
index 4c4c30a..6f8d9b8 100755
--- a/src/types/error.rs
+++ b/src/types/error.rs
@@ -30,6 +30,9 @@ pub enum Error {
         message: String,
     },
 
+    #[snafu(display("pool deletion is blocked for tenant '{}': {}", name, message))]
+    PoolDeleteBlocked { name: String, message: String },
+
     #[snafu(display("invalid tenant name '{}': {}", name, reason))]
     InvalidTenantName { name: String, reason: String },
 
diff --git a/src/types/v1alpha1/status.rs b/src/types/v1alpha1/status.rs
index 7139b37..d3276ef 100755
--- a/src/types/v1alpha1/status.rs
+++ b/src/types/v1alpha1/status.rs
@@ -18,8 +18,150 @@ pub mod state;
 use schemars::JsonSchema;
 use serde::{Deserialize, Serialize};
 
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+pub enum ConditionType {
+    Ready,
+    Reconciling,
+    Degraded,
+    SpecValid,
+    CredentialsReady,
+    KmsReady,
+    PoolsReady,
+    WorkloadsReady,
+}
+
+impl ConditionType {
+    pub const fn as_str(self) -> &'static str {
+        match self {
+            Self::Ready => "Ready",
+            Self::Reconciling => "Reconciling",
+            Self::Degraded => "Degraded",
+            Self::SpecValid => "SpecValid",
+            Self::CredentialsReady => "CredentialsReady",
+            Self::KmsReady => "KmsReady",
+            Self::PoolsReady => "PoolsReady",
+            Self::WorkloadsReady => "WorkloadsReady",
+        }
+    }
+
+    fn priority(type_: &str) -> Option<usize> {
+        [
+            Self::Ready,
+            Self::Reconciling,
+            Self::Degraded,
+            Self::SpecValid,
+            Self::CredentialsReady,
+            Self::KmsReady,
+            Self::PoolsReady,
+            Self::WorkloadsReady,
+        ]
+        .iter()
+        .position(|condition_type| condition_type.as_str() == type_)
+    }
+}
+
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+pub enum ConditionStatus {
+    True,
+    False,
+    Unknown,
+}
+
+impl ConditionStatus {
+    pub const fn as_str(self) -> &'static str {
+        match self {
+            Self::True => "True",
+            Self::False => "False",
+            Self::Unknown => "Unknown",
+        }
+    }
+}
+
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+pub enum CurrentState {
+    Ready,
+    Reconciling,
+    Blocked,
+    Degraded,
+    NotReady,
+    Unknown,
+}
+
+impl CurrentState {
+    pub const fn as_str(self) -> &'static str {
+        match self {
+            Self::Ready => "Ready",
+            Self::Reconciling => "Reconciling",
+            Self::Blocked => "Blocked",
+            Self::Degraded => "Degraded",
+            Self::NotReady => "NotReady",
+            Self::Unknown => "Unknown",
+        }
+    }
+}
+
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+pub enum Reason {
+    ReconcileStarted,
+    ReconcileSucceeded,
+    InvalidTenantName,
+    ImmutableFieldModified,
+    CredentialSecretNotFound,
+    CredentialSecretMissingKey,
+    CredentialSecretInvalidEncoding,
+    CredentialSecretTooShort,
+    KmsSecretNotFound,
+    KmsSecretMissingKey,
+    KmsConfigInvalid,
+    PoolDeleteBlocked,
+    StatefulSetApplyFailed,
+    StatefulSetUpdateValidationFailed,
+    RolloutInProgress,
+    PodsNotReady,
+    PoolDegraded,
+    KubernetesApiError,
+    StatusPatchFailed,
+    ObservedGenerationStale,
+}
+
+impl Reason {
+    pub const fn as_str(self) -> &'static str {
+        match self {
+            Self::ReconcileStarted => "ReconcileStarted",
+            Self::ReconcileSucceeded => "ReconcileSucceeded",
+            Self::InvalidTenantName => "InvalidTenantName",
+            Self::ImmutableFieldModified => "ImmutableFieldModified",
+            Self::CredentialSecretNotFound => "CredentialSecretNotFound",
+            Self::CredentialSecretMissingKey => "CredentialSecretMissingKey",
+            Self::CredentialSecretInvalidEncoding => "CredentialSecretInvalidEncoding",
+            Self::CredentialSecretTooShort => "CredentialSecretTooShort",
+            Self::KmsSecretNotFound => "KmsSecretNotFound",
+            Self::KmsSecretMissingKey => "KmsSecretMissingKey",
+            Self::KmsConfigInvalid => "KmsConfigInvalid",
+            Self::PoolDeleteBlocked => "PoolDeleteBlocked",
+            Self::StatefulSetApplyFailed => "StatefulSetApplyFailed",
+            Self::StatefulSetUpdateValidationFailed => "StatefulSetUpdateValidationFailed",
+            Self::RolloutInProgress => "RolloutInProgress",
+            Self::PodsNotReady => "PodsNotReady",
+            Self::PoolDegraded => "PoolDegraded",
+            Self::KubernetesApiError => "KubernetesApiError",
+            Self::StatusPatchFailed => "StatusPatchFailed",
+            Self::ObservedGenerationStale => "ObservedGenerationStale",
+        }
+    }
+}
+
+pub struct ConditionInput {
+    pub type_: ConditionType,
+    pub status: ConditionStatus,
+    pub reason: Reason,
+    pub message: String,
+    pub observed_generation: Option<i64>,
+    pub now: String,
+}
+
 /// Kubernetes standard condition for Tenant resources
-#[derive(Deserialize, Serialize, Clone, Debug, JsonSchema)]
+#[derive(Deserialize, Serialize, Clone, Debug, JsonSchema, PartialEq)]
 #[serde(rename_all = "camelCase")]
 pub struct Condition {
     /// Type of condition (Ready, Progressing, Degraded)
@@ -44,7 +186,7 @@ pub struct Condition {
     pub message: String,
 }
 
-#[derive(Deserialize, Serialize, Clone, Debug, JsonSchema, Default)]
+#[derive(Deserialize, Serialize, Clone, Debug, JsonSchema, Default, PartialEq)]
 #[serde(rename_all = "camelCase")]
 pub struct Status {
     pub current_state: String,
@@ -62,3 +204,285 @@ pub struct Status {
     pub conditions: Vec<Condition>,
     // pub certificates: certificate::Status,
 }
+
+impl Status {
+    pub fn upsert_condition(&mut self, input: ConditionInput) {
+        let type_ = input.type_.as_str();
+        if let Some(condition) = self
+            .conditions
+            .iter_mut()
+            .find(|condition| condition.type_ == type_)
+        {
+            if condition.status != input.status.as_str() {
+                condition.last_transition_time = Some(input.now);
+            }
+            condition.status = input.status.as_str().to_string();
+            condition.reason = input.reason.as_str().to_string();
+            condition.message = input.message;
+            condition.observed_generation = input.observed_generation;
+        } else {
+            self.conditions.push(Condition {
+                type_: type_.to_string(),
+                status: input.status.as_str().to_string(),
+                last_transition_time: Some(input.now),
+                observed_generation: input.observed_generation,
+                reason: input.reason.as_str().to_string(),
+                message: input.message,
+            });
+        }
+        self.sort_conditions();
+    }
+
+    pub fn condition(&self, type_: ConditionType) -> Option<&Condition> {
+        self.condition_by_type(type_.as_str())
+    }
+
+    pub fn condition_by_type(&self, type_: &str) -> Option<&Condition> {
+        self.conditions.iter().find(|condition| {
+            condition.type_ == type_ && condition_matches_observed_generation(self, condition)
+        })
+    }
+
+    pub fn sort_conditions(&mut self) {
+        self.conditions.sort_by(|left, right| {
+            match (
+                ConditionType::priority(&left.type_),
+                ConditionType::priority(&right.type_),
+            ) {
+                (Some(left), Some(right)) => left.cmp(&right),
+                (Some(_), None) => std::cmp::Ordering::Less,
+                (None, Some(_)) => std::cmp::Ordering::Greater,
+                (None, None) => left.type_.cmp(&right.type_),
+            }
+        });
+    }
+
+    pub fn condition_is_true(&self, type_: ConditionType) -> bool {
+        self.condition(type_)
+            .is_some_and(|condition| condition.status == ConditionStatus::True.as_str())
+    }
+
+    pub fn condition_is_false(&self, type_: ConditionType) -> bool {
+        self.condition(type_)
+            .is_some_and(|condition| condition.status == ConditionStatus::False.as_str())
+    }
+}
+
+pub fn canonical_state(state: Option<&str>) -> String {
+    canonical_known_state(state)
+        .unwrap_or(CurrentState::Unknown.as_str())
+        .to_string()
+}
+
+pub fn canonical_filter_state(state: Option<&str>) -> Option<String> {
+    canonical_known_state(state).map(ToOwned::to_owned)
+}
+
+fn canonical_known_state(state: Option<&str>) -> Option<&'static str> {
+    let normalized = state?
+        .trim()
+        .to_ascii_lowercase()
+        .replace([' ', '_', '-'], "");
+    match normalized.as_str() {
+        "ready" | "initialized" | "running" => Some(CurrentState::Ready.as_str()),
+        "reconciling" | "updating" | "creating" | "progressing" | "provisioning" => {
+            Some(CurrentState::Reconciling.as_str())
+        }
+        "blocked" => Some(CurrentState::Blocked.as_str()),
+        "degraded" => Some(CurrentState::Degraded.as_str()),
+        "notready" | "failed" | "error" => Some(CurrentState::NotReady.as_str()),
+        "unknown" | "stopped" => Some(CurrentState::Unknown.as_str()),
+        _ => None,
+    }
+}
+
+pub fn summarize_current_state(status: &Status) -> String {
+    if status.condition_is_true(ConditionType::Ready)
+        && !status.condition_is_true(ConditionType::Degraded)
+        && !status.condition_is_true(ConditionType::Reconciling)
+    {
+        return CurrentState::Ready.as_str().to_string();
+    }
+
+    if has_blocked_condition(status) {
+        return CurrentState::Blocked.as_str().to_string();
+    }
+
+    if status.condition_is_true(ConditionType::Reconciling) {
+        return CurrentState::Reconciling.as_str().to_string();
+    }
+
+    if status.condition_is_true(ConditionType::Degraded) {
+        return CurrentState::Degraded.as_str().to_string();
+    }
+
+    if status.condition_is_false(ConditionType::Ready) {
+        return CurrentState::NotReady.as_str().to_string();
+    }
+
+    CurrentState::Unknown.as_str().to_string()
+}
+
+pub fn primary_condition(status: &Status) -> Option<&Condition> {
+    status
+        .conditions
+        .iter()
+        .find(|condition| {
+            condition_matches_observed_generation(status, condition)
+                && condition.status == ConditionStatus::False.as_str()
+                && is_blocked_reason(&condition.reason)
+        })
+        .or_else(|| {
+            status.conditions.iter().find(|condition| {
+                condition_matches_observed_generation(status, condition)
+                    && condition.type_ == ConditionType::Degraded.as_str()
+                    && condition.status == ConditionStatus::True.as_str()
+            })
+        })
+        .or_else(|| {
+            status.conditions.iter().find(|condition| {
+                condition_matches_observed_generation(status, condition)
+                    && condition.type_ == ConditionType::Reconciling.as_str()
+                    && condition.status == ConditionStatus::True.as_str()
+            })
+        })
+        .or_else(|| {
+            status.conditions.iter().find(|condition| {
+                condition_matches_observed_generation(status, condition)
+                    && condition.type_ == ConditionType::Ready.as_str()
+                    && condition.status == ConditionStatus::False.as_str()
+            })
+        })
+}
+
+pub fn is_blocked_reason(reason: &str) -> bool {
+    matches!(
+        reason,
+        "InvalidTenantName"
+            | "ImmutableFieldModified"
+            | "CredentialSecretNotFound"
+            | "CredentialSecretMissingKey"
+            | "CredentialSecretInvalidEncoding"
+            | "CredentialSecretTooShort"
+            | "KmsSecretNotFound"
+            | "KmsSecretMissingKey"
+            | "KmsConfigInvalid"
+            | "PoolDeleteBlocked"
+            | "StatefulSetUpdateValidationFailed"
+    )
+}
+
+fn has_blocked_condition(status: &Status) -> bool {
+    status.conditions.iter().any(|condition| {
+        condition_matches_observed_generation(status, condition)
+            && condition.status == ConditionStatus::False.as_str()
+            && is_blocked_reason(&condition.reason)
+    })
+}
+
+fn condition_matches_observed_generation(status: &Status, condition: &Condition) -> bool {
+    match (status.observed_generation, condition.observed_generation) {
+        (Some(status_generation), Some(condition_generation)) => {
+            condition_generation == status_generation
+        }
+        _ => true,
+    }
+}
+
+pub fn next_actions_for_reason(reason: &str) -> Vec<&'static str> {
+    match reason {
+        "CredentialSecretNotFound" => vec!["createCredentialSecret"],
+        "CredentialSecretMissingKey" => vec!["addRequiredSecretKey"],
+        "CredentialSecretInvalidEncoding" => vec!["replaceSecretValueWithUtf8"],
+        "CredentialSecretTooShort" => vec!["rotateCredentialSecret"],
+        "KmsSecretNotFound" => vec!["createKmsSecret"],
+        "KmsSecretMissingKey" => vec!["addRequiredKmsSecretKey"],
+        "KmsConfigInvalid" => vec!["fixKmsConfig"],
+        "InvalidTenantName" => vec!["renameTenant"],
+        "ImmutableFieldModified" => vec!["restoreImmutableField"],
+        "PoolDeleteBlocked" => vec!["restorePoolSpec", "startDecommissionAfterRestore"],
+        "StatefulSetUpdateValidationFailed" => vec!["restoreImmutableField"],
+        "StatefulSetApplyFailed" => vec!["retry", "inspectOperatorLogs"],
+        "RolloutInProgress" => vec!["waitForRollout"],
+        "PodsNotReady" => vec!["inspectPods", "inspectEvents"],
+        "PoolDegraded" => vec![
+            "inspectPools",
+            "inspectPods",
+            "inspectEvents",
+            "inspectOperatorLogs",
+        ],
+        "KubernetesApiError" => vec!["retry", "inspectOperatorLogs"],
+        "ObservedGenerationStale" => vec!["waitForReconcile"],
+        _ => Vec::new(),
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn summary_ignores_conditions_from_older_observed_generation() {
+        let status = Status {
+            current_state: "Blocked".to_string(),
+            observed_generation: Some(2),
+            conditions: vec![
+                condition(
+                    "CredentialsReady",
+                    "False",
+                    "CredentialSecretNotFound",
+                    Some(1),
+                ),
+                condition("Reconciling", "True", "KubernetesApiError", Some(2)),
+                condition("Ready", "Unknown", "KubernetesApiError", Some(2)),
+            ],
+            ..Default::default()
+        };
+
+        assert_eq!(summarize_current_state(&status), "Reconciling");
+        assert_eq!(
+            primary_condition(&status).map(|condition| condition.reason.as_str()),
+            Some("KubernetesApiError")
+        );
+    }
+
+    #[test]
+    fn canonical_state_accepts_case_insensitive_aliases() {
+        assert_eq!(canonical_state(Some("ready")), "Ready");
+        assert_eq!(canonical_state(Some("Updating")), "Reconciling");
+        assert_eq!(canonical_state(Some("creating")), "Reconciling");
+    }
+
+    #[test]
+    fn next_actions_cover_degraded_and_stale_statuses() {
+        assert_eq!(
+            next_actions_for_reason("PoolDegraded"),
+            vec![
+                "inspectPools",
+                "inspectPods",
+                "inspectEvents",
+                "inspectOperatorLogs"
+            ]
+        );
+        assert_eq!(
+            next_actions_for_reason("ObservedGenerationStale"),
+            vec!["waitForReconcile"]
+        );
+    }
+
+    fn condition(
+        type_: &str,
+        status: &str,
+        reason: &str,
+        observed_generation: Option<i64>,
+    ) -> Condition {
+        Condition {
+            type_: type_.to_string(),
+            status: status.to_string(),
+            last_transition_time: Some("now".to_string()),
+            observed_generation,
+            reason: reason.to_string(),
+            message: reason.to_string(),
+        }
+    }
+}
diff --git a/src/types/v1alpha1/status/pool.rs b/src/types/v1alpha1/status/pool.rs
index d76158a..21c8c1f 100755
--- a/src/types/v1alpha1/status/pool.rs
+++ b/src/types/v1alpha1/status/pool.rs
@@ -18,7 +18,7 @@ use serde::{Deserialize, Serialize};
 use std::borrow::Cow;
 use strum::Display;
 
-#[derive(Deserialize, Serialize, Clone, Debug, KubeSchema)]
+#[derive(Deserialize, Serialize, Clone, Debug, KubeSchema, PartialEq)]
 #[serde(rename_all = "camelCase")]
 pub struct Pool {
     /// Name of the StatefulSet for this pool
@@ -56,7 +56,7 @@ pub struct Pool {
     pub last_update_time: Option<String>,
 }
 
-#[derive(Deserialize, Serialize, Clone, Debug, Display)]
+#[derive(Deserialize, Serialize, Clone, Debug, Display, PartialEq, Eq)]
 pub enum PoolState {
     #[strum(to_string = "PoolCreated")]
     Created,
diff --git a/src/types/v1alpha1/tenant.rs b/src/types/v1alpha1/tenant.rs
index 779aa39..9dd9356 100755
--- a/src/types/v1alpha1/tenant.rs
+++ b/src/types/v1alpha1/tenant.rs
@@ -263,23 +263,42 @@ impl Tenant {
         let current_revision = status.and_then(|s| s.current_revision.clone());
         let update_revision = status.and_then(|s| s.update_revision.clone());
 
-        // Determine pool state based on StatefulSet status
+        // Determine pool state based on StatefulSet status. Kubernetes StatefulSet
+        // status is authoritative only after the controller has observed the latest
+        // generation; revision mismatch also means a rollout is still in progress.
         let state = if let Some(status) = status {
-            let desired = status.replicas;
+            let desired = ss
+                .spec
+                .as_ref()
+                .and_then(|spec| spec.replicas)
+                .unwrap_or(status.replicas);
             let ready = status.ready_replicas.unwrap_or(0);
             let updated = status.updated_replicas.unwrap_or(0);
             let current = status.current_replicas.unwrap_or(0);
+            let observed_current = match (status.observed_generation, ss.metadata.generation) {
+                (Some(observed), Some(generation)) => observed >= generation,
+                (Some(_), None) | (None, None) => true,
+                (None, Some(_)) => false,
+            };
+            let revisions_match = match (&status.current_revision, &status.update_revision) {
+                (Some(current_revision), Some(update_revision)) => {
+                    current_revision == update_revision
+                }
+                (None, None) => true,
+                _ => false,
+            };
 
             if desired == 0 {
                 PoolState::NotCreated
+            } else if !observed_current
+                || !revisions_match
+                || updated < desired
+                || current < desired
+            {
+                PoolState::Updating
             } else if ready == desired && updated == desired {
-                // All replicas are ready and updated
                 PoolState::RolloutComplete
-            } else if updated < desired || current < desired {
-                // Rollout in progress
-                PoolState::Updating
             } else if ready < desired {
-                // Some replicas not ready
                 PoolState::Degraded
             } else {
                 PoolState::Initialized
@@ -360,6 +379,62 @@ pub fn validate_dns1035_label(name: &str) -> Result<(), types::error::Error> {
 
 #[cfg(test)]
 mod tests {
+    use crate::types::v1alpha1::status::pool::PoolState;
+    use k8s_openapi::api::apps::v1::{StatefulSet, StatefulSetSpec, StatefulSetStatus};
+    use k8s_openapi::apimachinery::pkg::apis::meta::v1::ObjectMeta;
+
+    fn statefulset_with_status(
+        generation: i64,
+        observed_generation: i64,
+        replicas: i32,
+        ready_replicas: i32,
+        updated_replicas: i32,
+        current_revision: &str,
+        update_revision: &str,
+    ) -> StatefulSet {
+        StatefulSet {
+            metadata: ObjectMeta {
+                name: Some("test-tenant-pool-0".to_string()),
+                generation: Some(generation),
+                ..Default::default()
+            },
+            spec: Some(StatefulSetSpec {
+                replicas: Some(replicas),
+                ..Default::default()
+            }),
+            status: Some(StatefulSetStatus {
+                observed_generation: Some(observed_generation),
+                replicas,
+                ready_replicas: Some(ready_replicas),
+                updated_replicas: Some(updated_replicas),
+                current_replicas: Some(replicas),
+                current_revision: Some(current_revision.to_string()),
+                update_revision: Some(update_revision.to_string()),
+                ..Default::default()
+            }),
+            ..Default::default()
+        }
+    }
+
+    #[test]
+    fn pool_status_treats_stale_statefulset_observation_as_updating() {
+        let tenant = crate::tests::create_test_tenant(None, None);
+        let ss = statefulset_with_status(2, 1, 4, 4, 4, "rev-a", "rev-a");
+
+        let pool_status = tenant.build_pool_status("pool-0", &ss);
+
+        assert_eq!(pool_status.state, PoolState::Updating);
+    }
+
+    #[test]
+    fn pool_status_requires_current_and_update_revisions_to_match() {
+        let tenant = crate::tests::create_test_tenant(None, None);
+        let ss = statefulset_with_status(2, 2, 4, 4, 4, "rev-a", "rev-b");
+
+        let pool_status = tenant.build_pool_status("pool-0", &ss);
+
+        assert_eq!(pool_status.state, PoolState::Updating);
+    }
 
     // Test 1: Default behavior - no custom SA
     #[test]