diff --git a/src/cluster/mod.rs b/src/cluster/mod.rs
index a72bd63..2bf16da 100644
--- a/src/cluster/mod.rs
+++ b/src/cluster/mod.rs
@@ -1,12 +1,13 @@
 pub mod kubeconfig;
 
-use futures::TryStreamExt as _;
+use std::time::Duration;
+
 use gpui_kit::{App, BorrowAppContext as _, Global};
-use k8s_openapi::api::core::v1::Namespace;
-use kube::api::{ListParams, WatchEvent, WatchParams};
-use kube::{Api, Client, Config};
+use kube::{Client, Config};
 use tokio::sync::watch;
 
+use crate::models::{NamespaceScope, ResourceKind, SnapshotPhase};
+
 /// One selectable cluster context: a kubectl context or a cluster added to krusty.
 #[derive(Clone)]
 pub struct ContextInfo {
@@ -45,6 +46,9 @@ pub struct ClusterState {
     namespaces: Vec<String>,
     context_epoch: u64,
     namespace_cancel: Option<watch::Sender<bool>>,
+    /// Kept alive for the app lifetime: the selector's handle on the shared
+    /// namespaces watcher that the Namespaces page also reads.
+    namespace_subscription: Option<crate::repository::Subscription>,
 }
 
 impl Global for ClusterState {}
@@ -149,22 +153,14 @@ pub fn init(cx: &mut App) {
         .or_else(|| contexts.first().map(|context| context.id.clone()));
 
     tracing::info!(preferred = ?preferred, "cluster: selected startup context");
-    let client = preferred.as_deref().and_then(|id| {
-        let context = contexts.iter().find(|context| context.id == id)?;
-        match client_for(&tokio, kubeconfig.as_ref(), &context.source, &store) {
-            Ok(client) => Some(client),
-            Err(error) => {
-                tracing::warn!(%error, context = %context.id, "cluster: startup client failed");
-                None
-            }
-        }
-    });
+    let startup = preferred
+        .as_deref()
+        .and_then(|id| contexts.iter().find(|context| context.id == id).cloned());
 
-    let connected = client.is_some();
     cx.set_global(ClusterState {
         _runtime: runtime,
         tokio,
-        client,
+        client: None,
         kubeconfig,
         contexts,
         active: preferred,
@@ -173,142 +169,73 @@ pub fn init(cx: &mut App) {
         namespaces: Vec::new(),
         context_epoch: 0,
         namespace_cancel: None,
+        namespace_subscription: None,
     });
-    refresh_namespaces(cx);
+    // Building the client can run an auth plugin (kubelogin, gcloud, …):
+    // keep it off the UI thread so startup stays instant.
+    if let Some(context) = startup {
+        spawn_client_build(cx, context);
+    }
     tracing::info!(
         elapsed_ms = started.elapsed().as_millis() as u64,
-        connected,
         "cluster: init complete"
     );
 }
 
-async fn run_namespace_watcher(
-    client: Client,
-    tx: watch::Sender<Vec<String>>,
-    mut cancel: watch::Receiver<bool>,
-) {
-    let api: Api<Namespace> = Api::all(client);
-    let list_params = ListParams::default();
-    let watch_params = WatchParams::default().timeout(280);
-
-    loop {
-        if cancel.has_changed().is_err() || *cancel.borrow() {
-            return;
-        }
-
-        let listed = tokio::select! {
-            _ = cancel.changed() => return,
-            result = api.list(&list_params) => result,
-        };
-
-        let list = match listed {
-            Ok(list) => list,
-            Err(error) => {
-                tracing::warn!(%error, "cluster: namespace list failed, retrying");
-                tokio::select! {
-                    _ = cancel.changed() => return,
-                    _ = tokio::time::sleep(std::time::Duration::from_millis(1500)) => continue,
-                };
-            }
-        };
-
-        let mut names = crate::repository::namespace_names(list.items);
-        let mut resource_version = list.metadata.resource_version.unwrap_or_default();
-        let _ = tx.send(names.clone());
-
-        let watched = tokio::select! {
-            _ = cancel.changed() => return,
-            result = api.watch(&watch_params, &resource_version) => result,
-        };
-
-        let mut stream = match watched {
-            Ok(stream) => Box::pin(stream),
-            Err(error) => {
-                tracing::warn!(%error, "cluster: namespace watch setup failed, retrying");
-                tokio::select! {
-                    _ = cancel.changed() => return,
-                    _ = tokio::time::sleep(std::time::Duration::from_millis(1500)) => continue,
-                };
-            }
-        };
-
-        loop {
-            let event = tokio::select! {
-                _ = cancel.changed() => return,
-                item = stream.try_next() => item,
-            };
-
-            match event {
-                Ok(Some(WatchEvent::Added(ns))) => {
-                    if let Some(name) = ns.metadata.name
-                        && !names.contains(&name)
-                    {
-                        names.push(name);
-                        names.sort_unstable();
-                        let _ = tx.send(names.clone());
-                    }
-                    if let Some(version) = ns.metadata.resource_version {
-                        resource_version = version;
-                    }
-                }
-                Ok(Some(WatchEvent::Deleted(ns))) => {
-                    if let Some(name) = ns.metadata.name
-                        && let Some(idx) = names.iter().position(|n| n == &name)
-                    {
-                        names.remove(idx);
-                        let _ = tx.send(names.clone());
-                    }
-                    if let Some(version) = ns.metadata.resource_version {
-                        resource_version = version;
-                    }
-                }
-                Ok(Some(WatchEvent::Bookmark(bm))) => {
-                    resource_version = bm.metadata.resource_version;
-                }
-                Ok(Some(WatchEvent::Modified(_))) => {}
-                Ok(Some(WatchEvent::Error(err))) => {
-                    tracing::warn!(%err, "cluster: namespace watch error, relisting");
-                    break;
-                }
-                Ok(None) => {
-                    break;
-                }
-                Err(error) => {
-                    tracing::warn!(%error, "cluster: namespace watch stream error, relisting");
-                    break;
-                }
-            }
-        }
-    }
-}
-
-/// Start or restart the continuous namespace list/watch loop.
+/// Start or restart the shared namespace subscription. The selector and the
+/// Namespaces resource page read from the same repository watcher, so
+/// namespaces are fetched once per context and follow the saved poll rate.
 pub fn refresh_namespaces(cx: &mut App) {
-    let state = cx.global::<ClusterState>();
-    let Some(client) = state.client.clone() else {
+    if cx.global::<ClusterState>().client.is_none() {
         return;
-    };
-    let tokio = state.tokio.clone();
-
-    if let Some(cancel) = &state.namespace_cancel {
+    }
+    // Stop the previous pump and release the previous handle; dropping the
+    // last handle cancels the shared watcher task.
+    if let Some(cancel) = &cx.global::<ClusterState>().namespace_cancel {
         let _ = cancel.send(true);
     }
 
-    let (cancel_tx, cancel_rx) = watch::channel(false);
-    let (names_tx, mut names_rx) = watch::channel(state.namespaces.clone());
+    let subscription =
+        match crate::repository::subscribe(cx, ResourceKind::Namespaces, NamespaceScope::All) {
+            Ok(subscription) => subscription,
+            Err(error) => {
+                tracing::debug!(%error, "cluster: namespace subscription unavailable");
+                return;
+            }
+        };
 
+    let (cancel_tx, mut cancel_rx) = watch::channel(false);
     cx.update_global::<ClusterState, _>(|state, _| {
         state.namespace_cancel = Some(cancel_tx);
+        state.namespace_subscription = Some(subscription.clone());
     });
 
-    tokio.spawn(run_namespace_watcher(client, names_tx, cancel_rx));
-
+    let mut subscription = subscription;
     cx.spawn(async move |cx: &mut gpui_kit::AsyncApp| {
-        while names_rx.changed().await.is_ok() {
-            let names = names_rx.borrow_and_update().clone();
+        loop {
+            let snapshot = tokio::select! {
+                _ = cancel_rx.changed() => break,
+                snapshot = subscription.next() => snapshot,
+            };
+            let Some(snapshot) = snapshot else {
+                break;
+            };
+            let names: Vec<String> = snapshot
+                .items()
+                .iter()
+                .map(|item| item.name.clone())
+                .collect();
             cx.update_global::<ClusterState, _>(|state, _| {
-                state.namespaces = names.clone();
-                state.reachable = Some(true);
+                // An empty list means the watcher has not loaded yet (or the
+                // context is switching); keep the previous selector contents.
+                if !names.is_empty() {
+                    state.namespaces = names;
+                }
+                match snapshot.phase() {
+                    SnapshotPhase::Ready => state.reachable = Some(true),
+                    SnapshotPhase::Error(_) => state.reachable = Some(false),
+                    SnapshotPhase::Loading => {}
+                }
             });
         }
     })
@@ -350,19 +277,21 @@ fn build_contexts(
     contexts
 }
 
-fn client_for(
-    tokio: &tokio::runtime::Handle,
-    kubeconfig: Option<&kube::config::Kubeconfig>,
-    source: &ContextSource,
-    store: &kubeconfig::ConfigStore,
+/// Build a kube client without blocking the caller. Auth plugins (kubelogin,
+/// gcloud, aws, …) can take seconds, so this always runs on the cluster
+/// runtime, never the UI thread.
+async fn build_client(
+    kubeconfig: Option<kube::config::Kubeconfig>,
+    source: ContextSource,
+    store: kubeconfig::ConfigStore,
 ) -> Result<Client, &'static str> {
     let started = std::time::Instant::now();
-    let source_label = match source {
+    let source_label = match &source {
         ContextSource::Kubeconfig(name) => format!("kube:{name}"),
         ContextSource::App(id) => format!("app:{id}"),
     };
     tracing::debug!(context = %source_label, "cluster: building kube client");
-    let config = match source {
+    let mut config = match &source {
         ContextSource::Kubeconfig(name) => {
             let kubeconfig = kubeconfig.ok_or("kubeconfig unavailable")?;
             let opts = kube::config::KubeConfigOptions {
@@ -370,8 +299,8 @@ fn client_for(
                 cluster: None,
                 user: None,
             };
-            tokio
-                .block_on(Config::from_custom_kubeconfig(kubeconfig.clone(), &opts))
+            Config::from_custom_kubeconfig(kubeconfig, &opts)
+                .await
                 .map_err(|_| "invalid kubeconfig context")?
         }
         ContextSource::App(id) => {
@@ -380,12 +309,13 @@ fn client_for(
                 .iter()
                 .find(|entry| entry.id == *id)
                 .ok_or("unknown cluster")?;
-            load_inline(tokio, &entry.kubeconfig_yaml)?
+            load_inline(&entry.kubeconfig_yaml).await?
         }
     };
-    let client = tokio
-        .block_on(async { Client::try_from(config) })
-        .map_err(|_| "connection failed")?;
+    // kube defaults to 30s; fail dead endpoints faster and let the retry
+    // policy back off instead of stalling the UI.
+    config.connect_timeout = Some(Duration::from_secs(10));
+    let client = Client::try_from(config).map_err(|_| "connection failed")?;
     tracing::debug!(
         context = %source_label,
         elapsed_ms = started.elapsed().as_millis() as u64,
@@ -394,7 +324,62 @@ fn client_for(
     Ok(client)
 }
 
+/// Build the client for `context` in the background and publish it when ready.
+/// The epoch bump resubscribes every resource view; namespace polling starts
+/// here too. Stale builds (superseded by a newer switch) are discarded.
+fn spawn_client_build(cx: &mut App, context: ContextInfo) {
+    let (kubeconfig, store, tokio) = {
+        let state = cx.global::<ClusterState>();
+        (
+            state.kubeconfig.clone(),
+            state.store.clone(),
+            state.tokio.clone(),
+        )
+    };
+    let context_id = context.id.clone();
+    let started = std::time::Instant::now();
+    let building = tokio.spawn(build_client(kubeconfig, context.source, store));
+    cx.spawn(async move |cx: &mut gpui_kit::AsyncApp| {
+        let built = building.await;
+        cx.update(|cx| {
+            let state = cx.global::<ClusterState>();
+            if state.active_id() != Some(context_id.as_str()) {
+                tracing::debug!(context = %context_id, "cluster: discarding stale client build");
+                return;
+            }
+            match built {
+                Ok(Ok(client)) => {
+                    tracing::info!(
+                        context = %context_id,
+                        elapsed_ms = started.elapsed().as_millis() as u64,
+                        "cluster: client ready"
+                    );
+                    cx.update_global::<ClusterState, _>(|state, _| {
+                        state.client = Some(client);
+                        state.context_epoch = state.context_epoch.wrapping_add(1);
+                    });
+                    refresh_namespaces(cx);
+                }
+                Ok(Err(error)) => {
+                    tracing::warn!(%error, context = %context_id, "cluster: client build failed");
+                    cx.update_global::<ClusterState, _>(|state, _| {
+                        state.reachable = Some(false);
+                    });
+                }
+                Err(error) => {
+                    tracing::error!(%error, context = %context_id, "cluster: client build task failed");
+                    cx.update_global::<ClusterState, _>(|state, _| {
+                        state.reachable = Some(false);
+                    });
+                }
+            }
+        });
+    })
+    .detach();
+}
+
 /// Switch the active cluster context. Selection is remembered across launches.
+/// The new client is built in the background; views resubscribe when it lands.
 pub fn switch_context(cx: &mut App, id: &str) -> Result<(), &'static str> {
     let started = std::time::Instant::now();
     tracing::info!(context = %id, "cluster: switching context");
@@ -405,39 +390,37 @@ pub fn switch_context(cx: &mut App, id: &str) -> Result<(), &'static str> {
             .find(|context| context.id == id)
             .cloned()
             .ok_or("unknown context")?;
-        let client = client_for(
-            &state.tokio,
-            state.kubeconfig.as_ref(),
-            &context.source,
-            &state.store,
-        )?;
-        state.client = Some(client);
-        state.active = Some(context.id);
+        state.client = None;
+        state.active = Some(context.id.clone());
         state.reachable = None;
         state.namespaces.clear();
         state.context_epoch = state.context_epoch.wrapping_add(1);
         if let Some(cancel) = state.namespace_cancel.take() {
             let _ = cancel.send(true);
         }
+        state.namespace_subscription = None;
         state.store.set_last_context(Some(id));
         let _ = state.store.save();
-        Ok(())
+        Ok(context)
     });
-    if result.is_ok() {
-        refresh_namespaces(cx);
-    }
-    match &result {
-        Ok(()) => tracing::info!(
-            elapsed_ms = started.elapsed().as_millis() as u64,
-            "cluster: context switched"
-        ),
-        Err(error) => tracing::error!(
-            %error,
-            elapsed_ms = started.elapsed().as_millis() as u64,
-            "cluster: context switch failed"
-        ),
+    match result {
+        Ok(context) => {
+            spawn_client_build(cx, context);
+            tracing::info!(
+                elapsed_ms = started.elapsed().as_millis() as u64,
+                "cluster: context switched, client building"
+            );
+            Ok(())
+        }
+        Err(error) => {
+            tracing::error!(
+                %error,
+                elapsed_ms = started.elapsed().as_millis() as u64,
+                "cluster: context switch failed"
+            );
+            Err(error)
+        }
     }
-    result
 }
 
 fn rebuild_contexts(state: &mut ClusterState) {
@@ -451,14 +434,11 @@ fn rebuild_contexts(state: &mut ClusterState) {
     }
 }
 
-fn load_inline(tokio: &tokio::runtime::Handle, yaml: &str) -> Result<Config, &'static str> {
+async fn load_inline(yaml: &str) -> Result<Config, &'static str> {
     let kubeconfig: kube::config::Kubeconfig =
         serde_saphyr::from_str(yaml).map_err(|_| "invalid kubeconfig")?;
-    tokio
-        .block_on(Config::from_custom_kubeconfig(
-            kubeconfig,
-            &kube::config::KubeConfigOptions::default(),
-        ))
+    Config::from_custom_kubeconfig(kubeconfig, &kube::config::KubeConfigOptions::default())
+        .await
         .map_err(|_| "invalid kubeconfig")
 }
 
@@ -559,20 +539,19 @@ users:
         assert_eq!(contexts[0].label(), "kind-dev");
 
         let runtime = tokio::runtime::Runtime::new().expect("runtime");
-        let client = client_for(runtime.handle(), None, &contexts[0].source, &store);
+        let client = runtime.block_on(build_client(None, contexts[0].source.clone(), store));
         assert!(client.is_ok());
     }
 
     #[test]
     fn parses_inline_kubeconfig() {
         let runtime = tokio::runtime::Runtime::new().expect("runtime");
-        let handle = runtime.handle().clone();
         let yaml = kube::config::Kubeconfig::read()
             .ok()
             .and_then(|kubeconfig| serde_saphyr::to_string(&kubeconfig).ok());
         let Some(yaml) = yaml else {
             return;
         };
-        assert!(load_inline(&handle, &yaml).is_ok());
+        assert!(runtime.block_on(load_inline(&yaml)).is_ok());
     }
 }
diff --git a/src/models/mod.rs b/src/models/mod.rs
index 78f3c34..9cc2ec6 100644
--- a/src/models/mod.rs
+++ b/src/models/mod.rs
@@ -8,6 +8,6 @@ pub use cluster::NamespaceScope;
 pub use resources::{
     ConfigMapsModel, DaemonSetsModel, DeploymentsModel, IngressesModel, Item, JobsModel,
     NodesModel, ObjectKey, PersistentVolumeClaimsModel, PersistentVolumesModel, PodsModel,
-    ResourceDetail, ResourceDetailState, ResourceDetailUpdate, ResourceKind, ResourceSnapshot,
-    ResourceStore, SecretsModel, ServicesModel, SnapshotPhase,
+    PollRate, ResourceDetail, ResourceDetailState, ResourceDetailUpdate, ResourceKind,
+    ResourceSnapshot, ResourceStore, SecretsModel, ServicesModel, SnapshotPhase, WatchMode,
 };
diff --git a/src/models/resources/mod.rs b/src/models/resources/mod.rs
index 0272b37..a63ad08 100644
--- a/src/models/resources/mod.rs
+++ b/src/models/resources/mod.rs
@@ -67,4 +67,4 @@ pub use service_accounts::ServiceAccountsModel;
 pub use services::ServicesModel;
 pub use stateful_sets::StatefulSetsModel;
 pub use storage_classes::StorageClassesModel;
-pub use store::{ObjectKey, ResourceSnapshot, ResourceStore, SnapshotPhase};
+pub use store::{ObjectKey, PollRate, ResourceSnapshot, ResourceStore, SnapshotPhase, WatchMode};
diff --git a/src/models/resources/store.rs b/src/models/resources/store.rs
index 4a18f4e..ecdda16 100644
--- a/src/models/resources/store.rs
+++ b/src/models/resources/store.rs
@@ -35,6 +35,7 @@ pub struct ResourceSnapshot {
     items: Arc<[Arc<Item>]>,
     revision: u64,
     phase: SnapshotPhase,
+    mode: WatchMode,
 }
 
 #[derive(Clone, Debug, PartialEq, Eq)]
@@ -44,12 +45,43 @@ pub enum SnapshotPhase {
     Error(String),
 }
 
+/// How the snapshot is kept up to date. `Poll` carries the cadence so the UI
+/// can show and adjust it.
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub enum WatchMode {
+    /// Live watch stream.
+    Watch,
+    /// Watch is not permitted; the store is re-listed on this cadence.
+    Poll(PollRate),
+}
+
+/// Automatic re-list cadence while polling. `Off` disables the timer; a
+/// manual refresh still re-lists once. Persisted per resource kind.
+#[derive(Clone, Copy, Debug, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
+pub enum PollRate {
+    /// No automatic polling.
+    Off,
+    /// Poll every this many seconds.
+    Every(u64),
+}
+
+impl PollRate {
+    pub const DEFAULT: PollRate = PollRate::Every(5);
+}
+
+impl Default for PollRate {
+    fn default() -> Self {
+        Self::DEFAULT
+    }
+}
+
 impl ResourceSnapshot {
     pub fn loading() -> Self {
         Self {
             items: Arc::from(Vec::<Arc<Item>>::new()),
             revision: 0,
             phase: SnapshotPhase::Loading,
+            mode: WatchMode::Watch,
         }
     }
 
@@ -58,9 +90,15 @@ impl ResourceSnapshot {
             items: Arc::from(Vec::<Arc<Item>>::new()),
             revision: 0,
             phase: SnapshotPhase::Error(message.into()),
+            mode: WatchMode::Watch,
         }
     }
 
+    pub fn with_mode(mut self, mode: WatchMode) -> Self {
+        self.mode = mode;
+        self
+    }
+
     pub fn items(&self) -> &[Arc<Item>] {
         &self.items
     }
@@ -72,6 +110,10 @@ impl ResourceSnapshot {
     pub fn phase(&self) -> &SnapshotPhase {
         &self.phase
     }
+
+    pub fn mode(&self) -> WatchMode {
+        self.mode
+    }
 }
 
 /// Accumulates watch events; the repository flushes snapshots at a bounded rate.
@@ -148,6 +190,7 @@ impl ResourceStore {
             items: self.objects.values().cloned().collect(),
             revision: self.revision,
             phase,
+            mode: WatchMode::Watch,
         }
     }
 }
@@ -229,4 +272,15 @@ mod tests {
         assert_eq!(snapshot.items().len(), 1);
         assert_eq!(snapshot.items()[0].name, "b");
     }
+
+    #[test]
+    fn snapshots_carry_the_watch_mode() {
+        let mut store = ResourceStore::new();
+        let snapshot = store.snapshot(SnapshotPhase::Ready);
+        assert_eq!(snapshot.mode(), WatchMode::Watch);
+        let snapshot = snapshot.with_mode(WatchMode::Poll(PollRate::Every(5)));
+        assert_eq!(snapshot.mode(), WatchMode::Poll(PollRate::Every(5)));
+        assert_eq!(ResourceSnapshot::loading().mode(), WatchMode::Watch);
+        assert_eq!(ResourceSnapshot::failed("x").mode(), WatchMode::Watch);
+    }
 }
diff --git a/src/repository/mod.rs b/src/repository/mod.rs
index 92767c8..1b9a79f 100644
--- a/src/repository/mod.rs
+++ b/src/repository/mod.rs
@@ -3,8 +3,10 @@
 
 pub mod custom_resources;
 pub mod resources;
+pub mod retry;
+pub mod watch_prefs;
 
-pub use resources::namespace_names;
+pub use watch_prefs::{WatchPreference, WatchPrefs};
 
 use std::collections::HashMap;
 use std::sync::{Arc, Weak};
@@ -13,7 +15,7 @@ use gpui_kit::{App, BorrowAppContext as _, Global};
 use tokio::sync::watch;
 
 use crate::cluster::ClusterState;
-use crate::models::{NamespaceScope, ResourceKind, ResourceSnapshot};
+use crate::models::{NamespaceScope, PollRate, ResourceKind, ResourceSnapshot, WatchMode};
 
 #[derive(Clone, Debug, PartialEq, Eq, Hash)]
 struct SubscriptionKey {
@@ -26,17 +28,21 @@ struct Entry {
     snapshots: watch::Sender<Arc<ResourceSnapshot>>,
     _cancel: watch::Sender<bool>,
     refresh: watch::Sender<u64>,
+    prefs: watch::Sender<WatchPrefs>,
 }
 
-#[derive(Default)]
 pub struct Repository {
     entries: HashMap<SubscriptionKey, Weak<Entry>>,
+    watch_prefs: watch_prefs::WatchPrefsStore,
 }
 
 impl Global for Repository {}
 
 pub fn init(cx: &mut App) {
-    cx.set_global(Repository::default());
+    cx.set_global(Repository {
+        entries: HashMap::new(),
+        watch_prefs: watch_prefs::WatchPrefsStore::load(),
+    });
 }
 
 /// A live list/watch subscription. Dropping the last handle for a key cancels
@@ -68,7 +74,6 @@ impl Subscription {
 
 /// Subscribe to a kind in a namespace scope. Subscriptions with the same key
 /// share one list/watch task, so opening the same resource twice costs nothing.
-#[allow(dead_code)]
 pub fn subscribe(
     cx: &mut App,
     kind: ResourceKind,
@@ -108,18 +113,76 @@ pub(crate) fn subscribe_kind(
             };
         }
 
-        let (snapshots, rx) = watch::channel(Arc::new(ResourceSnapshot::loading()));
+        let mut initial_prefs = repository.watch_prefs.prefs_for(kind);
+        if initial_prefs.mode == WatchPreference::Watch
+            && resources::watcher::watch_denied(key.epoch, kind)
+        {
+            // Watch is known to be denied for this kind: go straight to
+            // polling instead of re-trying the doomed watch on every
+            // namespace filter change. The saved preference stays `Watch`.
+            initial_prefs.mode = WatchPreference::Poll;
+        }
+        let initial_mode = match initial_prefs.mode {
+            WatchPreference::Watch => WatchMode::Watch,
+            WatchPreference::Poll => WatchMode::Poll(initial_prefs.rate),
+        };
+
+        let (snapshots, rx) = watch::channel(Arc::new(
+            ResourceSnapshot::loading().with_mode(initial_mode),
+        ));
         let (cancel, cancel_rx) = watch::channel(false);
         let (refresh, refresh_rx) = watch::channel(0u64);
+        let (prefs, prefs_rx) = watch::channel(initial_prefs);
         let entry = Arc::new(Entry {
             snapshots: snapshots.clone(),
             _cancel: cancel,
             refresh,
+            prefs,
         });
-        let watch_task =
-            resources::watcher::run(client, kind, scope, snapshots, cancel_rx, refresh_rx);
+        let watch_task = resources::watcher::run(
+            client, kind, scope, key.epoch, snapshots, cancel_rx, refresh_rx, prefs_rx,
+        );
         tokio.spawn(watch_task);
         repository.entries.insert(key, Arc::downgrade(&entry));
         Subscription { rx, _entry: entry }
     }))
 }
+
+/// Set whether a resource kind is watched live or polled. The saved poll rate
+/// is kept. Applies to every running watch task for the kind and is saved.
+pub fn set_watch_preference(cx: &mut App, kind: ResourceKind, mode: WatchPreference) {
+    if mode == WatchPreference::Watch {
+        // Explicit request: forget any RBAC denial so the watch is retried.
+        let epoch = cx.global::<ClusterState>().context_epoch();
+        resources::watcher::clear_watch_denied(epoch, kind);
+    }
+    update_watch_prefs(cx, kind, |prefs| prefs.mode = mode);
+}
+
+/// Set a kind's poll rate and switch it to poll mode. Applies to every
+/// running watch task for the kind and is saved.
+pub fn set_poll_rate(cx: &mut App, kind: ResourceKind, rate: PollRate) {
+    update_watch_prefs(cx, kind, |prefs| {
+        prefs.mode = WatchPreference::Poll;
+        prefs.rate = rate;
+    });
+}
+
+fn update_watch_prefs(cx: &mut App, kind: ResourceKind, change: impl FnOnce(&mut WatchPrefs)) {
+    let epoch = cx.global::<ClusterState>().context_epoch();
+    cx.update_global::<Repository, _>(|repository, _| {
+        repository.watch_prefs.update(kind, change);
+        if let Err(error) = repository.watch_prefs.save() {
+            tracing::warn!(%error, "repository: failed to save watch prefs");
+        }
+        let prefs = repository.watch_prefs.prefs_for(kind);
+        for (key, entry) in &repository.entries {
+            if key.kind == kind
+                && key.epoch == epoch
+                && let Some(entry) = entry.upgrade()
+            {
+                entry.prefs.send_replace(prefs);
+            }
+        }
+    });
+}
diff --git a/src/repository/resources/mod.rs b/src/repository/resources/mod.rs
index 25556a3..bf1748e 100644
--- a/src/repository/resources/mod.rs
+++ b/src/repository/resources/mod.rs
@@ -30,5 +30,3 @@ pub mod services;
 pub mod stateful_sets;
 pub mod storage_classes;
 pub mod watcher;
-
-pub use watcher::namespace_names;
diff --git a/src/repository/resources/pods/mod.rs b/src/repository/resources/pods/mod.rs
index b8115ef..c64162b 100644
--- a/src/repository/resources/pods/mod.rs
+++ b/src/repository/resources/pods/mod.rs
@@ -21,6 +21,7 @@ use crate::models::resources::pod::detail::{
 use crate::models::{NamespaceScope, ResourceKind};
 use gpui_kit::App;
 
+use crate::repository::retry::{Retry, RetryAction};
 use crate::repository::{Subscription, subscribe_kind};
 
 const METRICS_POLL_INTERVAL: Duration = Duration::from_secs(10);
@@ -113,41 +114,106 @@ async fn watch_logs(
     mut cancel: watch::Receiver<bool>,
 ) {
     let api: Api<Pod> = Api::namespaced(client, &namespace);
-    let params = LogParams {
-        container: Some(container),
-        follow,
-        tail_lines: Some(2_000),
-        ..LogParams::default()
-    };
-    let stream = tokio::select! {
-        _ = cancel.changed() => return,
-        result = api.log_stream(&name, &params) => result,
-    };
-    let stream = match stream {
-        Ok(stream) => stream,
-        Err(error) => {
-            let _ = updates.send(Err(format!("Could not stream logs: {error}")));
-            return;
-        }
-    };
-    let mut lines = stream.lines();
+    let mut retry = Retry::new();
+    // Timestamp of the last streamed line; reconnects resume after it instead
+    // of replaying the tail, so a dropped follow stream does not duplicate
+    // lines. kubectl does the same with --since-time.
+    let mut last_timestamp: Option<jiff::Timestamp> = None;
+
     loop {
-        tokio::select! {
+        let params = LogParams {
+            container: Some(container.clone()),
+            follow,
+            // Timestamps are only needed to resume a dropped follow stream.
+            timestamps: follow,
+            tail_lines: if last_timestamp.is_some() {
+                None
+            } else {
+                Some(2_000)
+            },
+            since_time: last_timestamp,
+            ..LogParams::default()
+        };
+        let timestamps = params.timestamps;
+        let stream = tokio::select! {
             _ = cancel.changed() => return,
-            line = lines.try_next() => {
-                match line {
-                    Ok(Some(line)) => {
-                        if updates.send(Ok(line.trim_end_matches('\r').to_string())).is_err() {
-                            return;
-                        }
+            result = api.log_stream(&name, &params) => result,
+        };
+        let mut lines = match stream {
+            Ok(stream) => {
+                retry.list_succeeded();
+                stream.lines()
+            }
+            Err(error) => {
+                if !follow {
+                    let _ = updates.send(Err(format!("Could not stream logs: {error}")));
+                    return;
+                }
+                match retry.failed(&error) {
+                    RetryAction::Retry(delay) => {
+                        tracing::warn!(%error, pod = %name, delay_ms = delay.as_millis() as u64, "pods: log stream setup failed, retrying");
+                        tokio::select! {
+                            _ = cancel.changed() => return,
+                            _ = tokio::time::sleep(delay) => continue,
+                        };
                     }
-                    Ok(None) => return,
-                    Err(error) => {
-                        let _ = updates.send(Err(format!("Log stream error: {error}")));
+                    RetryAction::GiveUp => {
+                        let _ = updates.send(Err(format!("Could not stream logs: {error}")));
                         return;
                     }
                 }
             }
+        };
+
+        loop {
+            let line = tokio::select! {
+                _ = cancel.changed() => return,
+                line = lines.try_next() => line,
+            };
+            match line {
+                Ok(Some(raw)) => {
+                    retry.watch_succeeded();
+                    let raw = raw.trim_end_matches('\r');
+                    let text = if timestamps {
+                        match raw.split_once(' ') {
+                            Some((stamp, text)) => {
+                                if let Ok(timestamp) = stamp.parse::<jiff::Timestamp>() {
+                                    last_timestamp = Some(timestamp);
+                                }
+                                text
+                            }
+                            None => raw,
+                        }
+                    } else {
+                        raw
+                    };
+                    if updates.send(Ok(text.to_string())).is_err() {
+                        return;
+                    }
+                }
+                Ok(None) => {
+                    tracing::debug!(pod = %name, "pods: log stream ended");
+                    break;
+                }
+                Err(error) => {
+                    tracing::warn!(%error, pod = %name, "pods: log stream read error");
+                    break;
+                }
+            }
+        }
+
+        if !follow {
+            return;
+        }
+        // The follow stream dropped: reconnect after a soft backoff, resuming
+        // after the last received line.
+        let RetryAction::Retry(delay) = retry.failed_soft() else {
+            return;
+        };
+        tracing::warn!(pod = %name, delay_ms = delay.as_millis() as u64, "pods: log stream interrupted, reconnecting");
+        tokio::select! {
+            _ = cancel.changed() => return,
+            _ = tokio::time::sleep(delay) => {}
         }
     }
 }
diff --git a/src/repository/resources/watcher.rs b/src/repository/resources/watcher.rs
index 1029ff7..2f7710b 100644
--- a/src/repository/resources/watcher.rs
+++ b/src/repository/resources/watcher.rs
@@ -4,8 +4,9 @@
 //! [`ResourceStore`] and only publishes immutable snapshots to the UI, throttled
 //! to `FLUSH_INTERVAL` or every `FLUSH_AT` events, whichever comes first.
 
+use std::collections::HashSet;
 use std::fmt::Debug;
-use std::sync::Arc;
+use std::sync::{Arc, Mutex, OnceLock};
 use std::time::Duration;
 
 use futures::TryStreamExt as _;
@@ -29,29 +30,72 @@ use serde::de::DeserializeOwned;
 use tokio::sync::watch;
 
 use crate::models::{
-    Item, NamespaceScope, ObjectKey, ResourceKind, ResourceSnapshot, ResourceStore, SnapshotPhase,
+    Item, NamespaceScope, ObjectKey, PollRate, ResourceKind, ResourceSnapshot, ResourceStore,
+    SnapshotPhase, WatchMode,
 };
+use crate::repository::retry::{self, RequestFailure, Retry, RetryAction};
+use crate::repository::watch_prefs::{WatchPreference, WatchPrefs};
 
 const FLUSH_INTERVAL: Duration = Duration::from_millis(50);
 const FLUSH_AT: usize = 256;
-const RETRY_DELAY: Duration = Duration::from_millis(750);
+
+/// List chunk size. Big kinds (thousands of pods/events) render their first
+/// chunk immediately instead of waiting for one huge response.
+const LIST_CHUNK_SIZE: u32 = 500;
+
+/// Kinds whose watch the API server denied (RBAC) for the current context
+/// epoch. New subscriptions for these kinds skip the doomed watch attempt and
+/// start polling right away, so changing the namespace filter does not reset
+/// polling back to a live-watch attempt.
+static WATCH_DENIED: OnceLock<Mutex<HashSet<(u64, ResourceKind)>>> = OnceLock::new();
+
+fn denied_registry() -> &'static Mutex<HashSet<(u64, ResourceKind)>> {
+    WATCH_DENIED.get_or_init(|| Mutex::new(HashSet::new()))
+}
+
+pub fn watch_denied(epoch: u64, kind: ResourceKind) -> bool {
+    denied_registry()
+        .lock()
+        .is_ok_and(|set| set.contains(&(epoch, kind)))
+}
+
+/// Forget a denial: the user explicitly asked for live watch again.
+pub fn clear_watch_denied(epoch: u64, kind: ResourceKind) {
+    if let Ok(mut set) = denied_registry().lock() {
+        set.remove(&(epoch, kind));
+    }
+}
+
+fn mark_watch_denied(epoch: u64, kind: ResourceKind) {
+    if let Ok(mut set) = denied_registry().lock() {
+        set.retain(|(stale_epoch, _)| *stale_epoch == epoch);
+        set.insert((epoch, kind));
+    }
+}
 
 /// Spawned once per subscription. Keep it generic over the store; the typed
 /// kind dispatch lives in `run`.
+#[allow(clippy::too_many_arguments)]
 pub async fn run(
     client: Client,
     kind: ResourceKind,
     scope: NamespaceScope,
+    epoch: u64,
     tx: watch::Sender<Arc<ResourceSnapshot>>,
     cancel: watch::Receiver<bool>,
     refresh: watch::Receiver<u64>,
+    prefs: watch::Receiver<WatchPrefs>,
 ) {
     let mut task = WatchTask {
         client,
+        kind,
+        epoch,
         tx,
         cancel,
         refresh,
+        prefs,
         store: ResourceStore::new(),
+        mode: WatchMode::Watch,
     };
     match kind {
         ResourceKind::Pods => task.run_namespaced::<Pod, _>(scope, pod_item).await,
@@ -142,21 +186,34 @@ pub async fn run(
     }
 }
 
-pub fn namespace_names(namespaces: Vec<Namespace>) -> Vec<String> {
-    let mut names: Vec<_> = namespaces
-        .into_iter()
-        .filter_map(|namespace| namespace.metadata.name)
-        .collect();
-    names.sort_unstable();
-    names
-}
-
 struct WatchTask {
     client: Client,
+    kind: ResourceKind,
+    epoch: u64,
     tx: watch::Sender<Arc<ResourceSnapshot>>,
     cancel: watch::Receiver<bool>,
     refresh: watch::Receiver<u64>,
+    prefs: watch::Receiver<WatchPrefs>,
     store: ResourceStore,
+    mode: WatchMode,
+}
+
+/// Why the poll loop exited.
+enum PollOutcome {
+    /// Subscription cancelled or listing denied: stop the task.
+    Stop,
+    /// The user switched this kind back to watch mode.
+    ResumeWatch,
+}
+
+/// Outcome of a chunked list.
+enum ListOutcome {
+    /// Full list published; watch from this resource version.
+    Listed(String),
+    /// Subscription cancelled mid-list.
+    Cancelled,
+    /// A chunk request failed or timed out; the caller decides whether to retry.
+    Failed(RequestFailure),
 }
 
 impl WatchTask {
@@ -201,10 +258,10 @@ impl WatchTask {
         K: Resource + Clone + Debug + DeserializeOwned + Send + 'static,
         F: Fn(&K) -> Item + Copy,
     {
-        let list_params = ListParams::default();
         let watch_params = WatchParams::default().timeout(280);
         let mut resource_version = String::new();
         let mut need_list = true;
+        let mut retry = Retry::new();
         let mut ticker = tokio::time::interval(FLUSH_INTERVAL);
         ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
         ticker.tick().await;
@@ -213,43 +270,74 @@ impl WatchTask {
             if self.cancel.has_changed().is_err() {
                 return;
             }
+            // The user prefers polling for this kind: skip the watch entirely.
+            if self.prefs.borrow().mode == WatchPreference::Poll {
+                match self.poll_loop(&api, map).await {
+                    PollOutcome::Stop => return,
+                    PollOutcome::ResumeWatch => {
+                        need_list = true;
+                        retry = Retry::new();
+                        continue;
+                    }
+                }
+            }
             if need_list {
-                let listed = tokio::select! {
-                    _ = self.cancel.changed() => return,
-                    result = api.list(&list_params) => result,
-                };
-                match listed {
-                    Ok(list) => {
-                        resource_version = list.metadata.resource_version.unwrap_or_default();
-                        self.store.reset(list.items.iter().map(map));
-                        self.publish(SnapshotPhase::Ready);
+                match self.list_progressively(&api, map).await {
+                    ListOutcome::Listed(version) => {
+                        retry.list_succeeded();
+                        resource_version = version;
                         need_list = false;
                     }
-                    Err(error) => {
-                        tracing::warn!(%error, ?api, "repository: list failed");
-                        self.publish(SnapshotPhase::Error(format!("list failed: {error}")));
-                        if !self.sleep(RETRY_DELAY).await {
+                    ListOutcome::Cancelled => return,
+                    ListOutcome::Failed(failure) => match failure.action(&mut retry) {
+                        RetryAction::Retry(delay) => {
+                            tracing::warn!(%failure, ?api, delay_ms = delay.as_millis() as u64, "repository: list failed, retrying");
+                            self.publish(SnapshotPhase::Error(format!("list failed: {failure}")));
+                            if !self.sleep(delay).await {
+                                return;
+                            }
+                            continue;
+                        }
+                        RetryAction::GiveUp => {
+                            tracing::error!(%failure, ?api, "repository: list rejected, stopping watch");
+                            self.publish(SnapshotPhase::Error(format!("list rejected: {failure}")));
                             return;
                         }
-                        continue;
-                    }
+                    },
                 }
             }
 
             let watched = tokio::select! {
                 _ = self.cancel.changed() => return,
-                result = api.watch(&watch_params, &resource_version) => result,
+                result = retry::bounded(api.watch(&watch_params, &resource_version)) => result,
             };
             let mut stream = match watched {
-                Ok(stream) => Box::pin(stream),
-                Err(error) => {
-                    tracing::warn!(%error, "repository: watch setup failed");
-                    if !self.sleep(RETRY_DELAY).await {
-                        return;
-                    }
-                    need_list = true;
-                    continue;
+                Ok(stream) => {
+                    clear_watch_denied(self.epoch, self.kind);
+                    Box::pin(stream)
                 }
+                Err(failure) => match failure.action(&mut retry) {
+                    RetryAction::Retry(delay) => {
+                        tracing::warn!(%failure, delay_ms = delay.as_millis() as u64, "repository: watch setup failed, retrying");
+                        if !self.sleep(delay).await {
+                            return;
+                        }
+                        need_list = true;
+                        continue;
+                    }
+                    RetryAction::GiveUp => {
+                        tracing::warn!(%failure, ?api, "repository: watch denied, polling instead");
+                        mark_watch_denied(self.epoch, self.kind);
+                        match self.poll_loop(&api, map).await {
+                            PollOutcome::Stop => return,
+                            PollOutcome::ResumeWatch => {
+                                need_list = true;
+                                retry = Retry::new();
+                                continue;
+                            }
+                        }
+                    }
+                },
             };
 
             let mut relist = false;
@@ -261,6 +349,21 @@ impl WatchTask {
                         relist = true;
                         break;
                     }
+                    _ = self.prefs.changed() => {
+                        let prefs = *self.prefs.borrow_and_update();
+                        if prefs.mode == WatchPreference::Poll {
+                            tracing::info!(?api, "repository: watch disabled by preference, polling instead");
+                            drop(stream);
+                            match self.poll_loop(&api, map).await {
+                                PollOutcome::Stop => return,
+                                PollOutcome::ResumeWatch => {
+                                    relist = true;
+                                    retry = Retry::new();
+                                    break;
+                                }
+                            }
+                        }
+                    }
                     _ = ticker.tick() => {
                         if self.store.pending() > 0 {
                             self.publish(SnapshotPhase::Ready);
@@ -269,6 +372,7 @@ impl WatchTask {
                     event = stream.try_next() => {
                         match event {
                             Ok(Some(WatchEvent::Added(object) | WatchEvent::Modified(object))) => {
+                                retry.watch_succeeded();
                                 let item = map(&object);
                                 if !item.resource_version.is_empty() {
                                     resource_version.clone_from(&item.resource_version);
@@ -276,6 +380,7 @@ impl WatchTask {
                                 self.store.upsert(item);
                             }
                             Ok(Some(WatchEvent::Deleted(object))) => {
+                                retry.watch_succeeded();
                                 if let Some(version) = object.resource_version() {
                                     resource_version = version;
                                 }
@@ -285,30 +390,71 @@ impl WatchTask {
                                 ));
                             }
                             Ok(Some(WatchEvent::Bookmark(bookmark))) => {
+                                retry.watch_succeeded();
                                 resource_version = bookmark.metadata.resource_version;
                             }
                             Ok(Some(WatchEvent::Error(status))) => {
-                                tracing::warn!(
-                                    code = status.code,
-                                    message = %status.message,
-                                    "repository: watch error event"
-                                );
                                 relist = status.code == 410
                                     || status.reason == "Expired"
                                     || status.reason == "Gone";
-                                break;
+                                match retry.failed_status(status.code) {
+                                    RetryAction::Retry(delay) => {
+                                        tracing::warn!(
+                                            code = status.code,
+                                            message = %status.message,
+                                            delay_ms = delay.as_millis() as u64,
+                                            "repository: watch error event, retrying"
+                                        );
+                                        if !self.sleep(delay).await {
+                                            return;
+                                        }
+                                        break;
+                                    }
+                                    RetryAction::GiveUp => {
+                                        tracing::warn!(
+                                            code = status.code,
+                                            message = %status.message,
+                                            "repository: watch denied, polling instead"
+                                        );
+                                        mark_watch_denied(self.epoch, self.kind);
+                                        drop(stream);
+                                        match self.poll_loop(&api, map).await {
+                                            PollOutcome::Stop => return,
+                                            PollOutcome::ResumeWatch => {
+                                                relist = true;
+                                                retry = Retry::new();
+                                                break;
+                                            }
+                                        }
+                                    }
+                                }
                             }
                             Ok(None) => {
                                 tracing::debug!("repository: watch stream ended, resuming");
                                 break;
                             }
-                            Err(error) => {
-                                tracing::warn!(%error, "repository: watch stream failed");
-                                if !self.sleep(RETRY_DELAY).await {
-                                    return;
+                            Err(error) => match retry.failed(&error) {
+                                RetryAction::Retry(delay) => {
+                                    tracing::warn!(%error, delay_ms = delay.as_millis() as u64, "repository: watch stream failed, retrying");
+                                    if !self.sleep(delay).await {
+                                        return;
+                                    }
+                                    break;
                                 }
-                                break;
-                            }
+                                RetryAction::GiveUp => {
+                                    tracing::warn!(%error, ?api, "repository: watch denied, polling instead");
+                                    mark_watch_denied(self.epoch, self.kind);
+                                    drop(stream);
+                                    match self.poll_loop(&api, map).await {
+                                        PollOutcome::Stop => return,
+                                        PollOutcome::ResumeWatch => {
+                                            relist = true;
+                                            retry = Retry::new();
+                                            break;
+                                        }
+                                    }
+                                }
+                            },
                         }
                         if self.store.pending() >= FLUSH_AT {
                             self.publish(SnapshotPhase::Ready);
@@ -329,7 +475,7 @@ impl WatchTask {
     }
 
     fn publish(&mut self, phase: SnapshotPhase) {
-        let snapshot = Arc::new(self.store.snapshot(phase));
+        let snapshot = Arc::new(self.store.snapshot(phase).with_mode(self.mode));
         tracing::trace!(
             objects = self.store.len(),
             revision = snapshot.revision(),
@@ -345,6 +491,125 @@ impl WatchTask {
             _ = tokio::time::sleep(duration) => true,
         }
     }
+
+    /// List every object in chunks, publishing each chunk so large kinds
+    /// render progressively instead of waiting for one huge response.
+    /// Returns the resource version to watch from.
+    async fn list_progressively<K, F>(&mut self, api: &Api<K>, map: F) -> ListOutcome
+    where
+        K: Resource + Clone + Debug + DeserializeOwned + Send + 'static,
+        F: Fn(&K) -> Item + Copy,
+    {
+        let mut params = ListParams::default().limit(LIST_CHUNK_SIZE);
+        let mut resource_version = String::new();
+        let mut first_chunk = true;
+        loop {
+            let listed = tokio::select! {
+                _ = self.cancel.changed() => return ListOutcome::Cancelled,
+                result = retry::bounded(api.list(&params)) => result,
+            };
+            let chunk = match listed {
+                Ok(chunk) => chunk,
+                Err(failure) => return ListOutcome::Failed(failure),
+            };
+            // Every chunk carries the version of the same list snapshot.
+            if let Some(version) = chunk.metadata.resource_version {
+                resource_version = version;
+            }
+            if first_chunk {
+                self.store.reset(chunk.items.iter().map(map));
+                first_chunk = false;
+            } else {
+                for object in &chunk.items {
+                    self.store.upsert(map(object));
+                }
+            }
+            self.publish(SnapshotPhase::Ready);
+            match chunk.metadata.continue_ {
+                Some(token) if !token.is_empty() => {
+                    params = params.continue_token(&token);
+                }
+                _ => break,
+            }
+        }
+        ListOutcome::Listed(resource_version)
+    }
+
+    /// Poll instead of watching — because watch was denied (RBAC) or the user
+    /// prefers polling for this kind. Runs until the subscription is
+    /// cancelled, listing is denied, or the user switches back to watch mode.
+    /// `PollRate::Off` disables the timer; manual refresh still re-lists.
+    async fn poll_loop<K, F>(&mut self, api: &Api<K>, map: F) -> PollOutcome
+    where
+        K: Resource + Clone + Debug + DeserializeOwned + Send + 'static,
+        F: Fn(&K) -> Item + Copy,
+    {
+        self.mode = WatchMode::Poll(self.prefs.borrow().rate);
+        // Reflect the mode switch in the UI immediately, not after the first
+        // poll roundtrip.
+        if self.store.len() > 0 {
+            self.publish(SnapshotPhase::Ready);
+        }
+        let mut retry = Retry::new();
+
+        loop {
+            if self.cancel.has_changed().is_err() {
+                return PollOutcome::Stop;
+            }
+
+            match self.list_progressively(api, map).await {
+                ListOutcome::Listed(_) => {
+                    retry.list_succeeded();
+                }
+                ListOutcome::Cancelled => return PollOutcome::Stop,
+                ListOutcome::Failed(failure) => match failure.action(&mut retry) {
+                    RetryAction::Retry(delay) => {
+                        tracing::warn!(%failure, ?api, delay_ms = delay.as_millis() as u64, "repository: poll list failed, retrying");
+                        self.publish(SnapshotPhase::Error(format!("list failed: {failure}")));
+                        if !self.sleep(delay).await {
+                            return PollOutcome::Stop;
+                        }
+                        continue;
+                    }
+                    RetryAction::GiveUp => {
+                        tracing::error!(%failure, ?api, "repository: poll list rejected, stopping watch");
+                        self.publish(SnapshotPhase::Error(format!("list rejected: {failure}")));
+                        return PollOutcome::Stop;
+                    }
+                },
+            }
+
+            // Wait for the next tick. Refresh or a preference change re-list
+            // immediately; `Off` waits without a timer.
+            let rate = self.prefs.borrow().rate;
+            tokio::select! {
+                _ = self.cancel.changed() => return PollOutcome::Stop,
+                _ = self.refresh.changed() => {
+                    self.refresh.borrow_and_update();
+                }
+                _ = self.prefs.changed() => {
+                    let prefs = *self.prefs.borrow_and_update();
+                    self.mode = WatchMode::Poll(prefs.rate);
+                    if prefs.mode == WatchPreference::Watch {
+                        self.mode = WatchMode::Watch;
+                        self.publish(SnapshotPhase::Ready);
+                        return PollOutcome::ResumeWatch;
+                    }
+                    // Show the new rate/off state instantly; the re-list
+                    // happens on the next loop turn.
+                    self.publish(SnapshotPhase::Ready);
+                }
+                _ = async {
+                        match rate {
+                            PollRate::Every(secs) => {
+                                tokio::time::sleep(Duration::from_secs(secs)).await
+                            }
+                            PollRate::Off => std::future::pending().await,
+                        }
+                    } => {}
+            }
+        }
+    }
 }
 
 fn base_item<K>(object: &K) -> Item
@@ -903,3 +1168,20 @@ fn node_item(node: &Node) -> Item {
         .unwrap_or_default();
     item
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn watch_denied_registry_roundtrip() {
+        // Unique epoch so the process-global registry is not disturbed.
+        let epoch = u64::MAX;
+        assert!(!watch_denied(epoch, ResourceKind::Pods));
+        mark_watch_denied(epoch, ResourceKind::Pods);
+        assert!(watch_denied(epoch, ResourceKind::Pods));
+        assert!(!watch_denied(epoch, ResourceKind::Nodes));
+        clear_watch_denied(epoch, ResourceKind::Pods);
+        assert!(!watch_denied(epoch, ResourceKind::Pods));
+    }
+}
diff --git a/src/repository/retry.rs b/src/repository/retry.rs
new file mode 100644
index 0000000..3a41a48
--- /dev/null
+++ b/src/repository/retry.rs
@@ -0,0 +1,245 @@
+//! Retry policy for Kubernetes list/watch loops.
+//!
+//! Hard failures (RBAC denials, missing resource types) are retried a couple
+//! of times and then stop the loop instead of hammering the API server.
+//! Soft failures (network blips, timeouts, 429, 5xx) retry forever with
+//! capped exponential backoff.
+
+use std::time::Duration;
+
+/// Total attempts for a hard failure before the watch loop gives up.
+const MAX_HARD_ATTEMPTS: u32 = 2;
+
+/// Delay between attempts for a hard failure.
+const HARD_RETRY_DELAY: Duration = Duration::from_millis(1500);
+
+/// First backoff step for soft failures; doubles on each consecutive failure.
+const SOFT_BACKOFF_BASE: Duration = Duration::from_millis(500);
+
+/// Longest backoff step for soft failures.
+const SOFT_BACKOFF_MAX: Duration = Duration::from_secs(30);
+
+/// What a watch loop should do after a failure.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum RetryAction {
+    /// Wait for the given duration, then retry.
+    Retry(Duration),
+    /// Permanent failure: stop the loop.
+    GiveUp,
+}
+
+/// Longest a single list/watch-establishment request may take before it feeds
+/// the retry policy. kube leaves `read_timeout` unset, so without this a
+/// stalled connection hangs until the OS gives up on the TCP connection
+/// (~40s+). Watch streams and log follows are unaffected.
+pub const REQUEST_TIMEOUT: Duration = Duration::from_secs(20);
+
+/// Run one API request with a timeout so a stalled connection fails fast.
+pub async fn bounded<T>(
+    request: impl std::future::Future<Output = Result<T, kube::Error>>,
+) -> Result<T, RequestFailure> {
+    match tokio::time::timeout(REQUEST_TIMEOUT, request).await {
+        Ok(Ok(value)) => Ok(value),
+        Ok(Err(error)) => Err(RequestFailure::Api(error)),
+        Err(_) => Err(RequestFailure::TimedOut),
+    }
+}
+
+/// A failed Kubernetes API request.
+#[derive(Debug)]
+pub enum RequestFailure {
+    /// The API server answered with an error status.
+    Api(kube::Error),
+    /// The request produced nothing within [`REQUEST_TIMEOUT`].
+    TimedOut,
+}
+
+impl RequestFailure {
+    /// Classify into the retry policy; timeouts are always soft.
+    pub fn action(&self, retry: &mut Retry) -> RetryAction {
+        match self {
+            Self::Api(error) => retry.failed(error),
+            Self::TimedOut => retry.failed_soft(),
+        }
+    }
+}
+
+impl std::fmt::Display for RequestFailure {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        match self {
+            Self::Api(error) => write!(f, "{error}"),
+            Self::TimedOut => write!(f, "no response for {}s", REQUEST_TIMEOUT.as_secs()),
+        }
+    }
+}
+
+/// True when the HTTP status will never succeed on retry for a list/watch.
+pub fn is_hard_status(code: u16) -> bool {
+    matches!(code, 400 | 401 | 403 | 404 | 405 | 422)
+}
+
+/// True when the error is a permanent API rejection (see [`is_hard_status`]).
+/// Everything else — network, TLS, timeouts, 429, 5xx — is soft.
+pub fn is_hard(error: &kube::Error) -> bool {
+    match error {
+        kube::Error::Api(status) => is_hard_status(status.code),
+        _ => false,
+    }
+}
+
+/// Tracks consecutive failures for one watch loop.
+#[derive(Default)]
+pub struct Retry {
+    soft_failures: u32,
+    hard_failures: u32,
+}
+
+impl Retry {
+    pub fn new() -> Self {
+        Self::default()
+    }
+
+    /// A list call succeeded: connectivity works, so the soft backoff resets.
+    /// Hard failures stay counted: a resource can be listable but not
+    /// watchable, and list success must not clear that signal.
+    pub fn list_succeeded(&mut self) {
+        self.soft_failures = 0;
+    }
+
+    /// The watch stream delivered a real event: fully healthy.
+    pub fn watch_succeeded(&mut self) {
+        self.soft_failures = 0;
+        self.hard_failures = 0;
+    }
+
+    /// Decide what to do after a call failed with `error`.
+    pub fn failed(&mut self, error: &kube::Error) -> RetryAction {
+        self.fail(is_hard(error))
+    }
+
+    /// Decide what to do after a watch `Error` event carrying only a status.
+    pub fn failed_status(&mut self, code: u16) -> RetryAction {
+        self.fail(is_hard_status(code))
+    }
+
+    /// A failure with nothing to classify (stream ended, read error): always
+    /// soft.
+    pub fn failed_soft(&mut self) -> RetryAction {
+        self.fail(false)
+    }
+
+    fn fail(&mut self, hard: bool) -> RetryAction {
+        if hard {
+            self.hard_failures = self.hard_failures.saturating_add(1);
+            if self.hard_failures >= MAX_HARD_ATTEMPTS {
+                return RetryAction::GiveUp;
+            }
+            return RetryAction::Retry(HARD_RETRY_DELAY);
+        }
+        self.soft_failures = self.soft_failures.saturating_add(1);
+        RetryAction::Retry(self.backoff())
+    }
+
+    fn backoff(&self) -> Duration {
+        let shift = self.soft_failures.saturating_sub(1).min(6);
+        (SOFT_BACKOFF_BASE * (1u32 << shift)).min(SOFT_BACKOFF_MAX)
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn hard_failures_give_up_after_two_attempts() {
+        let mut retry = Retry::new();
+        assert_eq!(
+            retry.failed_status(403),
+            RetryAction::Retry(HARD_RETRY_DELAY)
+        );
+        assert_eq!(retry.failed_status(403), RetryAction::GiveUp);
+    }
+
+    #[test]
+    fn list_success_does_not_clear_hard_failures() {
+        let mut retry = Retry::new();
+        retry.failed_status(403);
+        retry.list_succeeded();
+        // A resource can be listable but not watchable: the second 403
+        // still gives up instead of looping forever.
+        assert_eq!(retry.failed_status(403), RetryAction::GiveUp);
+    }
+
+    #[test]
+    fn watch_success_resets_everything() {
+        let mut retry = Retry::new();
+        retry.failed_status(403);
+        retry.watch_succeeded();
+        assert_eq!(
+            retry.failed_status(403),
+            RetryAction::Retry(HARD_RETRY_DELAY)
+        );
+    }
+
+    #[test]
+    fn soft_failures_back_off_exponentially_with_cap() {
+        let mut retry = Retry::new();
+        let mut delays = Vec::new();
+        for _ in 0..10 {
+            let RetryAction::Retry(delay) = retry.failed_status(500) else {
+                panic!("soft failures never give up");
+            };
+            delays.push(delay);
+        }
+        assert_eq!(
+            delays,
+            vec![
+                Duration::from_millis(500),
+                Duration::from_secs(1),
+                Duration::from_secs(2),
+                Duration::from_secs(4),
+                Duration::from_secs(8),
+                Duration::from_secs(16),
+                Duration::from_secs(30),
+                Duration::from_secs(30),
+                Duration::from_secs(30),
+                Duration::from_secs(30),
+            ]
+        );
+    }
+
+    #[test]
+    fn list_success_resets_soft_backoff() {
+        let mut retry = Retry::new();
+        retry.failed_status(500);
+        retry.failed_status(500);
+        retry.list_succeeded();
+        assert_eq!(
+            retry.failed_status(500),
+            RetryAction::Retry(Duration::from_millis(500))
+        );
+    }
+
+    #[test]
+    fn status_classification() {
+        for code in [400, 401, 403, 404, 405, 422] {
+            assert!(is_hard_status(code), "{code} should be hard");
+        }
+        for code in [410, 429, 500, 502, 503] {
+            assert!(!is_hard_status(code), "{code} should be soft");
+        }
+    }
+
+    #[test]
+    fn timeouts_are_soft_and_back_off() {
+        let mut retry = Retry::new();
+        assert_eq!(
+            RequestFailure::TimedOut.action(&mut retry),
+            RetryAction::Retry(Duration::from_millis(500))
+        );
+        assert_eq!(
+            RequestFailure::TimedOut.action(&mut retry),
+            RetryAction::Retry(Duration::from_secs(1))
+        );
+    }
+}
diff --git a/src/repository/watch_prefs.rs b/src/repository/watch_prefs.rs
new file mode 100644
index 0000000..89e2cc8
--- /dev/null
+++ b/src/repository/watch_prefs.rs
@@ -0,0 +1,254 @@
+//! Persisted per-resource-kind update preferences: watch vs poll, and the
+//! poll rate.
+//!
+//! Lives under the app config folder, e.g. `~/.config/krusty/poll-rates.json`.
+//! Kinds without an entry use [`WatchPrefs::default`]: live watch with a
+//! five-second poll fallback.
+
+use std::collections::HashMap;
+
+use serde::{Deserialize, Serialize};
+
+use crate::models::{PollRate, ResourceKind};
+
+/// How the user wants a resource kind kept up to date.
+#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
+pub enum WatchPreference {
+    /// Live watch stream; falls back to polling when the API denies watch.
+    #[default]
+    Watch,
+    /// Never watch; re-list on the configured poll rate.
+    Poll,
+}
+
+/// The full per-kind update preference.
+#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
+pub struct WatchPrefs {
+    #[serde(default)]
+    pub mode: WatchPreference,
+    #[serde(default)]
+    pub rate: PollRate,
+}
+
+#[derive(Serialize, Deserialize, Debug, Default)]
+struct Store {
+    version: u32,
+    #[serde(default)]
+    kinds: HashMap<String, WatchPrefs>,
+}
+
+#[derive(Debug, Default)]
+pub struct WatchPrefsStore {
+    kinds: HashMap<String, WatchPrefs>,
+}
+
+/// The factory preference for a kind. Namespaces default to polling: many
+/// clusters deny `watch` on them and they change rarely. Picking "Live watch"
+/// in the UI saves and sticks across launches.
+fn default_prefs_for(kind: ResourceKind) -> WatchPrefs {
+    match kind {
+        ResourceKind::Namespaces => WatchPrefs {
+            mode: WatchPreference::Poll,
+            rate: PollRate::DEFAULT,
+        },
+        _ => WatchPrefs::default(),
+    }
+}
+
+impl WatchPrefsStore {
+    /// Path under the OS app config dir, e.g. ~/.config/krusty/poll-rates.json.
+    pub fn path() -> Option<std::path::PathBuf> {
+        dirs::config_dir().map(|d| d.join("krusty").join("poll-rates.json"))
+    }
+
+    pub fn load() -> Self {
+        let Some(path) = Self::path() else {
+            tracing::debug!("watch prefs: no config directory");
+            return Self::default();
+        };
+        match std::fs::read(&path) {
+            Ok(bytes) => match serde_json::from_slice::<Store>(&bytes) {
+                Ok(store) => {
+                    tracing::debug!(
+                        path = %path.display(),
+                        kinds = store.kinds.len(),
+                        "watch prefs: loaded"
+                    );
+                    Self { kinds: store.kinds }
+                }
+                Err(error) => {
+                    tracing::warn!(
+                        path = %path.display(),
+                        %error,
+                        "watch prefs: invalid config file, ignoring"
+                    );
+                    Self::default()
+                }
+            },
+            Err(error) => {
+                tracing::debug!(
+                    path = %path.display(),
+                    %error,
+                    "watch prefs: no config file yet"
+                );
+                Self::default()
+            }
+        }
+    }
+
+    pub fn save(&self) -> Result<(), &'static str> {
+        let Some(path) = Self::path() else {
+            tracing::warn!("watch prefs: save failed, no config directory");
+            return Err("no config directory");
+        };
+        if let Some(parent) = path.parent() {
+            std::fs::create_dir_all(parent).map_err(|_| "failed to create config directory")?;
+        }
+        let store = Store {
+            version: 2,
+            kinds: self.kinds.clone(),
+        };
+        let bytes =
+            serde_json::to_vec_pretty(&store).map_err(|_| "failed to serialize watch prefs")?;
+        std::fs::write(&path, bytes).map_err(|_| "failed to write watch prefs")?;
+        tracing::debug!(path = %path.display(), kinds = self.kinds.len(), "watch prefs: saved");
+        Ok(())
+    }
+
+    /// The saved preference for a kind, or its factory default.
+    pub fn prefs_for(&self, kind: ResourceKind) -> WatchPrefs {
+        self.kinds
+            .get(kind.label())
+            .copied()
+            .unwrap_or_else(|| default_prefs_for(kind))
+    }
+
+    /// Change a kind's preference. Entries equal to the kind's factory
+    /// default are dropped.
+    pub fn update(&mut self, kind: ResourceKind, change: impl FnOnce(&mut WatchPrefs)) {
+        let mut prefs = self.prefs_for(kind);
+        change(&mut prefs);
+        if prefs == default_prefs_for(kind) {
+            self.kinds.remove(kind.label());
+        } else {
+            self.kinds.insert(kind.label().to_string(), prefs);
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn defaults_to_live_watch_with_five_second_fallback() {
+        let store = WatchPrefsStore::default();
+        assert_eq!(
+            store.prefs_for(ResourceKind::Pods),
+            WatchPrefs {
+                mode: WatchPreference::Watch,
+                rate: PollRate::Every(5),
+            }
+        );
+    }
+
+    #[test]
+    fn namespaces_default_to_polling_until_watch_is_chosen() {
+        let mut store = WatchPrefsStore::default();
+        assert_eq!(
+            store.prefs_for(ResourceKind::Namespaces),
+            WatchPrefs {
+                mode: WatchPreference::Poll,
+                rate: PollRate::Every(5),
+            }
+        );
+
+        // Choosing live watch is saved and sticks.
+        store.update(ResourceKind::Namespaces, |prefs| {
+            prefs.mode = WatchPreference::Watch;
+        });
+        assert_eq!(
+            store.prefs_for(ResourceKind::Namespaces).mode,
+            WatchPreference::Watch
+        );
+        assert!(store.kinds.contains_key("Namespaces"));
+
+        // Switching back to the factory default drops the entry.
+        store.update(ResourceKind::Namespaces, |prefs| {
+            prefs.mode = WatchPreference::Poll;
+        });
+        assert!(!store.kinds.contains_key("Namespaces"));
+    }
+
+    #[test]
+    fn remembers_mode_and_rate_independently() {
+        let mut store = WatchPrefsStore::default();
+        store.update(ResourceKind::Pods, |prefs| prefs.rate = PollRate::Every(30));
+        store.update(ResourceKind::Events, |prefs| {
+            prefs.mode = WatchPreference::Poll;
+            prefs.rate = PollRate::Off;
+        });
+
+        // A custom rate survives switching between watch and poll.
+        store.update(ResourceKind::Pods, |prefs| {
+            prefs.mode = WatchPreference::Poll
+        });
+        assert_eq!(
+            store.prefs_for(ResourceKind::Pods),
+            WatchPrefs {
+                mode: WatchPreference::Poll,
+                rate: PollRate::Every(30),
+            }
+        );
+        store.update(ResourceKind::Pods, |prefs| {
+            prefs.mode = WatchPreference::Watch;
+        });
+        assert_eq!(
+            store.prefs_for(ResourceKind::Pods),
+            WatchPrefs {
+                mode: WatchPreference::Watch,
+                rate: PollRate::Every(30),
+            }
+        );
+
+        assert_eq!(
+            store.prefs_for(ResourceKind::Events),
+            WatchPrefs {
+                mode: WatchPreference::Poll,
+                rate: PollRate::Off,
+            }
+        );
+        assert_eq!(store.prefs_for(ResourceKind::Nodes), WatchPrefs::default());
+    }
+
+    #[test]
+    fn resetting_to_default_drops_the_entry() {
+        let mut store = WatchPrefsStore::default();
+        store.update(ResourceKind::Pods, |prefs| prefs.rate = PollRate::Every(30));
+        store.update(ResourceKind::Pods, |prefs| prefs.rate = PollRate::DEFAULT);
+        assert!(!store.kinds.contains_key("Pods"));
+    }
+
+    #[test]
+    fn serializes_as_kind_labels() {
+        let mut store = WatchPrefsStore::default();
+        store.update(ResourceKind::Deployments, |prefs| {
+            prefs.mode = WatchPreference::Poll;
+            prefs.rate = PollRate::Every(10);
+        });
+        let json = serde_json::to_string(&Store {
+            version: 2,
+            kinds: store.kinds.clone(),
+        })
+        .expect("serialize");
+        assert!(json.contains("\"Deployments\""));
+        let parsed: Store = serde_json::from_str(&json).expect("deserialize");
+        assert_eq!(
+            parsed.kinds["Deployments"],
+            WatchPrefs {
+                mode: WatchPreference::Poll,
+                rate: PollRate::Every(10),
+            }
+        );
+    }
+}
diff --git a/src/ui/resources/list.rs b/src/ui/resources/list.rs
index 571ad86..ef19de1 100644
--- a/src/ui/resources/list.rs
+++ b/src/ui/resources/list.rs
@@ -109,6 +109,7 @@ impl ResourceListView {
     fn list(&self, cx: &Context<Self>) -> AnyElement {
         let phase = self.vm.phase(cx);
         let count = self.vm.item_count(cx);
+        let mode = self.vm.mode(cx);
 
         let status = match &phase {
             SnapshotPhase::Loading => div()
@@ -137,7 +138,9 @@ impl ResourceListView {
         if self.kind.is_namespaced() {
             header_left = header_left.child(render_namespace_filter(&self.namespace_filter, cx));
         }
-        let header_left = header_left.child(status);
+        let header_left = header_left
+            .child(status)
+            .child(super::watch_mode::render(mode, self.kind, cx));
 
         let table_el = self.vm.render_table(cx);
         let handle = cx.entity().clone();
diff --git a/src/ui/resources/mod.rs b/src/ui/resources/mod.rs
index a5742e4..18b9c98 100644
--- a/src/ui/resources/mod.rs
+++ b/src/ui/resources/mod.rs
@@ -26,6 +26,7 @@ pub mod service_accounts;
 pub mod services;
 pub mod stateful_sets;
 pub mod storage_classes;
+pub mod watch_mode;
 
 #[allow(clippy::too_many_arguments)]
 pub fn render_detail<T: 'static>(
diff --git a/src/ui/resources/node/mod.rs b/src/ui/resources/node/mod.rs
index c5e3d56..51a60b0 100644
--- a/src/ui/resources/node/mod.rs
+++ b/src/ui/resources/node/mod.rs
@@ -55,6 +55,7 @@ impl NodesView {
     fn header(&self, cx: &Context<Self>) -> impl IntoElement {
         let phase = self.vm.read(cx).phase(cx);
         let count = self.vm.read(cx).item_count(cx);
+        let mode = self.vm.read(cx).mode(cx);
 
         let status = match &phase {
             SnapshotPhase::Loading => div()
@@ -91,7 +92,12 @@ impl NodesView {
                             .font_weight(FontWeight::MEDIUM)
                             .child(SharedString::from("Nodes")),
                     )
-                    .child(status),
+                    .child(status)
+                    .child(super::watch_mode::render(
+                        mode,
+                        crate::models::ResourceKind::Nodes,
+                        cx,
+                    )),
             )
             .child(
                 Button::new("refresh-nodes")
diff --git a/src/ui/resources/pod/mod.rs b/src/ui/resources/pod/mod.rs
index 3315854..116f752 100644
--- a/src/ui/resources/pod/mod.rs
+++ b/src/ui/resources/pod/mod.rs
@@ -19,7 +19,7 @@ use gpui_kit::{
     Render, SharedString, Styled as _, Subscription, Window, div, px,
 };
 
-use crate::models::{NamespaceScope, SnapshotPhase};
+use crate::models::{NamespaceScope, ResourceKind, SnapshotPhase};
 use crate::ui::resources::namespace_filter::{
     NamespaceFilter, NamespaceFilterEvent, render as render_namespace_filter,
 };
@@ -156,6 +156,7 @@ impl PodsView {
     fn header(&self, cx: &Context<Self>) -> impl IntoElement {
         let phase = self.vm.read(cx).phase(cx);
         let count = self.vm.read(cx).item_count(cx);
+        let mode = self.vm.read(cx).mode(cx);
 
         let status = match &phase {
             SnapshotPhase::Loading => div()
@@ -193,7 +194,8 @@ impl PodsView {
                             .child(SharedString::from("Pods")),
                     )
                     .child(render_namespace_filter(&self.namespace_filter, cx))
-                    .child(status),
+                    .child(status)
+                    .child(super::watch_mode::render(mode, ResourceKind::Pods, cx)),
             )
             .child(
                 Button::new("refresh-pods")
diff --git a/src/ui/resources/watch_mode.rs b/src/ui/resources/watch_mode.rs
new file mode 100644
index 0000000..22490fe
--- /dev/null
+++ b/src/ui/resources/watch_mode.rs
@@ -0,0 +1,87 @@
+//! Update-mode indicator for resource list headers.
+//!
+//! Shows a small "Live" badge while the watch stream is healthy and "Polling"
+//! when the store is re-listed on a timer — either because the API server
+//! denied `watch` (RBAC) or because the user prefers polling. The dropdown
+//! switches the kind between live watch and a poll rate (or off), and the
+//! choice is saved per resource kind.
+
+use gpui_kit::base::h_flex;
+use gpui_kit::component::{
+    ActiveTheme as _, IconName, Sizable as _,
+    button::{Button, ButtonVariants as _},
+    menu::{DropdownMenu as _, PopupMenuItem},
+};
+use gpui_kit::{AnyElement, App, IntoElement, ParentElement as _, Styled as _, div, px};
+
+use crate::models::{PollRate, ResourceKind, WatchMode};
+use crate::repository::WatchPreference;
+
+/// Poll intervals offered in the rate menu, in seconds.
+const POLL_RATES: [u64; 5] = [2, 5, 10, 30, 60];
+
+pub(crate) fn render(mode: WatchMode, kind: ResourceKind, cx: &App) -> AnyElement {
+    let (color, label, button_label) = match mode {
+        WatchMode::Watch => (cx.theme().success, "Live".to_string(), "Live".to_string()),
+        WatchMode::Poll(PollRate::Every(secs)) => (
+            cx.theme().warning,
+            format!("Polling · {secs}s"),
+            format!("{secs}s"),
+        ),
+        WatchMode::Poll(PollRate::Off) => (
+            cx.theme().warning,
+            "Polling off".to_string(),
+            "Off".to_string(),
+        ),
+    };
+
+    let badge = h_flex()
+        .gap_1()
+        .items_center()
+        .child(div().w(px(6.)).h(px(6.)).rounded_full().bg(color))
+        .child(
+            div()
+                .text_xs()
+                .text_color(cx.theme().muted_foreground)
+                .child(label),
+        );
+
+    let mode_button = Button::new("watch-mode")
+        .ghost()
+        .small()
+        .label(button_label)
+        .icon(IconName::ChevronsUpDown)
+        .tooltip(
+            "How this resource type stays up to date. \
+             Watch needs the `watch` permission; otherwise krusty falls back \
+             to polling. Saved per resource type.",
+        )
+        .dropdown_menu(move |mut menu, _, _| {
+            menu = menu.item(
+                PopupMenuItem::new("Live watch")
+                    .checked(mode == WatchMode::Watch)
+                    .on_click(move |_, _, cx| {
+                        crate::repository::set_watch_preference(cx, kind, WatchPreference::Watch);
+                    }),
+            );
+            menu = menu.separator();
+            for secs in POLL_RATES {
+                menu = menu.item(
+                    PopupMenuItem::new(format!("Poll every {secs}s"))
+                        .checked(mode == WatchMode::Poll(PollRate::Every(secs)))
+                        .on_click(move |_, _, cx| {
+                            crate::repository::set_poll_rate(cx, kind, PollRate::Every(secs));
+                        }),
+                );
+            }
+            menu.separator().item(
+                PopupMenuItem::new("Polling off")
+                    .checked(mode == WatchMode::Poll(PollRate::Off))
+                    .on_click(move |_, _, cx| {
+                        crate::repository::set_poll_rate(cx, kind, PollRate::Off);
+                    }),
+            )
+        });
+
+    badge.child(mode_button).into_any_element()
+}
diff --git a/src/view_models/mod.rs b/src/view_models/mod.rs
index 48895a0..8dcb985 100644
--- a/src/view_models/mod.rs
+++ b/src/view_models/mod.rs
@@ -647,6 +647,40 @@ impl ResourceViewModel {
         }
     }
 
+    pub fn mode(&self, cx: &App) -> crate::models::WatchMode {
+        match self {
+            Self::Pods(vm) => vm.read(cx).mode(cx),
+            Self::Deployments(vm) => vm.read(cx).mode(cx),
+            Self::StatefulSets(vm) => vm.read(cx).mode(cx),
+            Self::ReplicaSets(vm) => vm.read(cx).mode(cx),
+            Self::CronJobs(vm) => vm.read(cx).mode(cx),
+            Self::ServiceAccounts(vm) => vm.read(cx).mode(cx),
+            Self::StorageClasses(vm) => vm.read(cx).mode(cx),
+            Self::Namespaces(vm) => vm.read(cx).mode(cx),
+            Self::DaemonSets(vm) => vm.read(cx).mode(cx),
+            Self::Jobs(vm) => vm.read(cx).mode(cx),
+            Self::Services(vm) => vm.read(cx).mode(cx),
+            Self::Ingresses(vm) => vm.read(cx).mode(cx),
+            Self::PersistentVolumes(vm) => vm.read(cx).mode(cx),
+            Self::PersistentVolumeClaims(vm) => vm.read(cx).mode(cx),
+            Self::ConfigMaps(vm) => vm.read(cx).mode(cx),
+            Self::Secrets(vm) => vm.read(cx).mode(cx),
+            Self::Nodes(vm) => vm.read(cx).mode(cx),
+            Self::ClusterRoles(vm) => vm.read(cx).mode(cx),
+            Self::EndpointSlices(vm) => vm.read(cx).mode(cx),
+            Self::Endpoints(vm) => vm.read(cx).mode(cx),
+            Self::Events(vm) => vm.read(cx).mode(cx),
+            Self::HorizontalPodAutoscalers(vm) => vm.read(cx).mode(cx),
+            Self::Leases(vm) => vm.read(cx).mode(cx),
+            Self::NetworkPolicies(vm) => vm.read(cx).mode(cx),
+            Self::PodDisruptionBudgets(vm) => vm.read(cx).mode(cx),
+            Self::ReplicationControllers(vm) => vm.read(cx).mode(cx),
+            Self::RoleBindings(vm) => vm.read(cx).mode(cx),
+            Self::Roles(vm) => vm.read(cx).mode(cx),
+            Self::Custom(vm) => vm.read(cx).mode(cx),
+        }
+    }
+
     pub fn item_count(&self, cx: &App) -> usize {
         match self {
             Self::Pods(vm) => vm.read(cx).item_count(cx),
diff --git a/src/view_models/resources/cluster_roles/mod.rs b/src/view_models/resources/cluster_roles/mod.rs
index a33acfc..a2b9130 100644
--- a/src/view_models/resources/cluster_roles/mod.rs
+++ b/src/view_models/resources/cluster_roles/mod.rs
@@ -13,7 +13,7 @@ use tokio::sync::watch;
 
 use crate::cluster::ClusterState;
 use crate::models::resources::cluster_roles::ClusterRolesModel;
-use crate::models::{Item, NamespaceScope, ResourceKind, SnapshotPhase};
+use crate::models::{Item, NamespaceScope, ResourceKind, SnapshotPhase, WatchMode};
 use crate::repository;
 
 pub struct ClusterRolesViewModel {
@@ -92,6 +92,10 @@ impl ClusterRolesViewModel {
         self.model.read(cx).phase().clone()
     }
 
+    pub fn mode(&self, cx: &App) -> WatchMode {
+        self.model.read(cx).snapshot().mode()
+    }
+
     pub fn item_count(&self, cx: &App) -> usize {
         self.model.read(cx).item_count()
     }
diff --git a/src/view_models/resources/config_maps/mod.rs b/src/view_models/resources/config_maps/mod.rs
index 3551568..49be129 100644
--- a/src/view_models/resources/config_maps/mod.rs
+++ b/src/view_models/resources/config_maps/mod.rs
@@ -13,7 +13,7 @@ use tokio::sync::watch;
 
 use crate::cluster::ClusterState;
 use crate::models::resources::config_maps::ConfigMapsModel;
-use crate::models::{Item, NamespaceScope, ResourceKind, SnapshotPhase};
+use crate::models::{Item, NamespaceScope, ResourceKind, SnapshotPhase, WatchMode};
 use crate::repository;
 
 pub struct ConfigMapsViewModel {
@@ -88,6 +88,10 @@ impl ConfigMapsViewModel {
         self.model.read(cx).phase().clone()
     }
 
+    pub fn mode(&self, cx: &App) -> WatchMode {
+        self.model.read(cx).snapshot().mode()
+    }
+
     pub fn item_count(&self, cx: &App) -> usize {
         self.model.read(cx).item_count()
     }
diff --git a/src/view_models/resources/cron_jobs/mod.rs b/src/view_models/resources/cron_jobs/mod.rs
index 055d381..7a948be 100644
--- a/src/view_models/resources/cron_jobs/mod.rs
+++ b/src/view_models/resources/cron_jobs/mod.rs
@@ -13,7 +13,7 @@ use tokio::sync::watch;
 
 use crate::cluster::ClusterState;
 use crate::models::resources::cron_jobs::CronJobsModel;
-use crate::models::{Item, NamespaceScope, ResourceKind, SnapshotPhase};
+use crate::models::{Item, NamespaceScope, ResourceKind, SnapshotPhase, WatchMode};
 use crate::repository;
 
 pub struct CronJobsViewModel {
@@ -87,6 +87,10 @@ impl CronJobsViewModel {
         self.model.read(cx).phase().clone()
     }
 
+    pub fn mode(&self, cx: &App) -> WatchMode {
+        self.model.read(cx).snapshot().mode()
+    }
+
     pub fn item_count(&self, cx: &App) -> usize {
         self.model.read(cx).item_count()
     }
diff --git a/src/view_models/resources/custom.rs b/src/view_models/resources/custom.rs
index 4a1278b..74ebb31 100644
--- a/src/view_models/resources/custom.rs
+++ b/src/view_models/resources/custom.rs
@@ -13,7 +13,7 @@ use tokio::sync::watch;
 
 use crate::cluster::ClusterState;
 use crate::models::resources::custom::CustomResourceModel;
-use crate::models::{Item, NamespaceScope, ResourceKind, SnapshotPhase};
+use crate::models::{Item, NamespaceScope, ResourceKind, SnapshotPhase, WatchMode};
 use crate::repository;
 
 pub struct CustomResourceViewModel {
@@ -94,6 +94,10 @@ impl CustomResourceViewModel {
         self.model.read(cx).phase().clone()
     }
 
+    pub fn mode(&self, cx: &App) -> WatchMode {
+        self.model.read(cx).snapshot().mode()
+    }
+
     pub fn item_count(&self, cx: &App) -> usize {
         self.model.read(cx).item_count()
     }
diff --git a/src/view_models/resources/daemon_sets/mod.rs b/src/view_models/resources/daemon_sets/mod.rs
index e24fce9..6d68e71 100644
--- a/src/view_models/resources/daemon_sets/mod.rs
+++ b/src/view_models/resources/daemon_sets/mod.rs
@@ -13,7 +13,7 @@ use tokio::sync::watch;
 
 use crate::cluster::ClusterState;
 use crate::models::resources::daemon_sets::DaemonSetsModel;
-use crate::models::{Item, NamespaceScope, ResourceKind, SnapshotPhase};
+use crate::models::{Item, NamespaceScope, ResourceKind, SnapshotPhase, WatchMode};
 use crate::repository;
 
 pub struct DaemonSetsViewModel {
@@ -88,6 +88,10 @@ impl DaemonSetsViewModel {
         self.model.read(cx).phase().clone()
     }
 
+    pub fn mode(&self, cx: &App) -> WatchMode {
+        self.model.read(cx).snapshot().mode()
+    }
+
     pub fn item_count(&self, cx: &App) -> usize {
         self.model.read(cx).item_count()
     }
diff --git a/src/view_models/resources/deployments/mod.rs b/src/view_models/resources/deployments/mod.rs
index 8d3bc34..e3fe0c4 100644
--- a/src/view_models/resources/deployments/mod.rs
+++ b/src/view_models/resources/deployments/mod.rs
@@ -17,7 +17,7 @@ use tokio::sync::watch;
 
 use crate::cluster::ClusterState;
 use crate::models::resources::deployments::DeploymentsModel;
-use crate::models::{Item, NamespaceScope, ResourceKind, SnapshotPhase};
+use crate::models::{Item, NamespaceScope, ResourceKind, SnapshotPhase, WatchMode};
 use crate::repository;
 
 pub struct DeploymentsViewModel {
@@ -96,6 +96,10 @@ impl DeploymentsViewModel {
         self.model.read(cx).phase().clone()
     }
 
+    pub fn mode(&self, cx: &App) -> WatchMode {
+        self.model.read(cx).snapshot().mode()
+    }
+
     pub fn item_count(&self, cx: &App) -> usize {
         self.model.read(cx).item_count()
     }
diff --git a/src/view_models/resources/endpoint_slices/mod.rs b/src/view_models/resources/endpoint_slices/mod.rs
index 79c1657..ec0a9e8 100644
--- a/src/view_models/resources/endpoint_slices/mod.rs
+++ b/src/view_models/resources/endpoint_slices/mod.rs
@@ -13,7 +13,7 @@ use tokio::sync::watch;
 
 use crate::cluster::ClusterState;
 use crate::models::resources::endpoint_slices::EndpointSlicesModel;
-use crate::models::{Item, NamespaceScope, ResourceKind, SnapshotPhase};
+use crate::models::{Item, NamespaceScope, ResourceKind, SnapshotPhase, WatchMode};
 use crate::repository;
 
 pub struct EndpointSlicesViewModel {
@@ -92,6 +92,10 @@ impl EndpointSlicesViewModel {
         self.model.read(cx).phase().clone()
     }
 
+    pub fn mode(&self, cx: &App) -> WatchMode {
+        self.model.read(cx).snapshot().mode()
+    }
+
     pub fn item_count(&self, cx: &App) -> usize {
         self.model.read(cx).item_count()
     }
diff --git a/src/view_models/resources/endpoints/mod.rs b/src/view_models/resources/endpoints/mod.rs
index 0974332..a1e4602 100644
--- a/src/view_models/resources/endpoints/mod.rs
+++ b/src/view_models/resources/endpoints/mod.rs
@@ -13,7 +13,7 @@ use tokio::sync::watch;
 
 use crate::cluster::ClusterState;
 use crate::models::resources::endpoints::EndpointsModel;
-use crate::models::{Item, NamespaceScope, ResourceKind, SnapshotPhase};
+use crate::models::{Item, NamespaceScope, ResourceKind, SnapshotPhase, WatchMode};
 use crate::repository;
 
 pub struct EndpointsViewModel {
@@ -87,6 +87,10 @@ impl EndpointsViewModel {
         self.model.read(cx).phase().clone()
     }
 
+    pub fn mode(&self, cx: &App) -> WatchMode {
+        self.model.read(cx).snapshot().mode()
+    }
+
     pub fn item_count(&self, cx: &App) -> usize {
         self.model.read(cx).item_count()
     }
diff --git a/src/view_models/resources/events/mod.rs b/src/view_models/resources/events/mod.rs
index 35b5c1a..e0561f3 100644
--- a/src/view_models/resources/events/mod.rs
+++ b/src/view_models/resources/events/mod.rs
@@ -13,7 +13,7 @@ use tokio::sync::watch;
 
 use crate::cluster::ClusterState;
 use crate::models::resources::events::EventsModel;
-use crate::models::{Item, NamespaceScope, ResourceKind, SnapshotPhase};
+use crate::models::{Item, NamespaceScope, ResourceKind, SnapshotPhase, WatchMode};
 use crate::repository;
 
 pub struct EventsViewModel {
@@ -87,6 +87,10 @@ impl EventsViewModel {
         self.model.read(cx).phase().clone()
     }
 
+    pub fn mode(&self, cx: &App) -> WatchMode {
+        self.model.read(cx).snapshot().mode()
+    }
+
     pub fn item_count(&self, cx: &App) -> usize {
         self.model.read(cx).item_count()
     }
diff --git a/src/view_models/resources/horizontal_pod_autoscalers/mod.rs b/src/view_models/resources/horizontal_pod_autoscalers/mod.rs
index d12cc09..90bcc0d 100644
--- a/src/view_models/resources/horizontal_pod_autoscalers/mod.rs
+++ b/src/view_models/resources/horizontal_pod_autoscalers/mod.rs
@@ -13,7 +13,7 @@ use tokio::sync::watch;
 
 use crate::cluster::ClusterState;
 use crate::models::resources::horizontal_pod_autoscalers::HorizontalPodAutoscalersModel;
-use crate::models::{Item, NamespaceScope, ResourceKind, SnapshotPhase};
+use crate::models::{Item, NamespaceScope, ResourceKind, SnapshotPhase, WatchMode};
 use crate::repository;
 
 pub struct HorizontalPodAutoscalersViewModel {
@@ -92,6 +92,10 @@ impl HorizontalPodAutoscalersViewModel {
         self.model.read(cx).phase().clone()
     }
 
+    pub fn mode(&self, cx: &App) -> WatchMode {
+        self.model.read(cx).snapshot().mode()
+    }
+
     pub fn item_count(&self, cx: &App) -> usize {
         self.model.read(cx).item_count()
     }
diff --git a/src/view_models/resources/ingresses/mod.rs b/src/view_models/resources/ingresses/mod.rs
index 2729ad6..edf0627 100644
--- a/src/view_models/resources/ingresses/mod.rs
+++ b/src/view_models/resources/ingresses/mod.rs
@@ -13,7 +13,7 @@ use tokio::sync::watch;
 
 use crate::cluster::ClusterState;
 use crate::models::resources::ingresses::IngressesModel;
-use crate::models::{Item, NamespaceScope, ResourceKind, SnapshotPhase};
+use crate::models::{Item, NamespaceScope, ResourceKind, SnapshotPhase, WatchMode};
 use crate::repository;
 
 pub struct IngressesViewModel {
@@ -87,6 +87,10 @@ impl IngressesViewModel {
         self.model.read(cx).phase().clone()
     }
 
+    pub fn mode(&self, cx: &App) -> WatchMode {
+        self.model.read(cx).snapshot().mode()
+    }
+
     pub fn item_count(&self, cx: &App) -> usize {
         self.model.read(cx).item_count()
     }
diff --git a/src/view_models/resources/jobs/mod.rs b/src/view_models/resources/jobs/mod.rs
index d475915..32a77f5 100644
--- a/src/view_models/resources/jobs/mod.rs
+++ b/src/view_models/resources/jobs/mod.rs
@@ -13,7 +13,7 @@ use tokio::sync::watch;
 
 use crate::cluster::ClusterState;
 use crate::models::resources::jobs::JobsModel;
-use crate::models::{Item, NamespaceScope, ResourceKind, SnapshotPhase};
+use crate::models::{Item, NamespaceScope, ResourceKind, SnapshotPhase, WatchMode};
 use crate::repository;
 
 pub struct JobsViewModel {
@@ -87,6 +87,10 @@ impl JobsViewModel {
         self.model.read(cx).phase().clone()
     }
 
+    pub fn mode(&self, cx: &App) -> WatchMode {
+        self.model.read(cx).snapshot().mode()
+    }
+
     pub fn item_count(&self, cx: &App) -> usize {
         self.model.read(cx).item_count()
     }
diff --git a/src/view_models/resources/leases/mod.rs b/src/view_models/resources/leases/mod.rs
index b3cb0bb..642c0b8 100644
--- a/src/view_models/resources/leases/mod.rs
+++ b/src/view_models/resources/leases/mod.rs
@@ -13,7 +13,7 @@ use tokio::sync::watch;
 
 use crate::cluster::ClusterState;
 use crate::models::resources::leases::LeasesModel;
-use crate::models::{Item, NamespaceScope, ResourceKind, SnapshotPhase};
+use crate::models::{Item, NamespaceScope, ResourceKind, SnapshotPhase, WatchMode};
 use crate::repository;
 
 pub struct LeasesViewModel {
@@ -87,6 +87,10 @@ impl LeasesViewModel {
         self.model.read(cx).phase().clone()
     }
 
+    pub fn mode(&self, cx: &App) -> WatchMode {
+        self.model.read(cx).snapshot().mode()
+    }
+
     pub fn item_count(&self, cx: &App) -> usize {
         self.model.read(cx).item_count()
     }
diff --git a/src/view_models/resources/namespaces/mod.rs b/src/view_models/resources/namespaces/mod.rs
index 00558ad..f1d864f 100644
--- a/src/view_models/resources/namespaces/mod.rs
+++ b/src/view_models/resources/namespaces/mod.rs
@@ -13,7 +13,7 @@ use tokio::sync::watch;
 
 use crate::cluster::ClusterState;
 use crate::models::resources::namespaces::NamespacesModel;
-use crate::models::{Item, NamespaceScope, ResourceKind, SnapshotPhase};
+use crate::models::{Item, NamespaceScope, ResourceKind, SnapshotPhase, WatchMode};
 use crate::repository;
 
 pub struct NamespacesViewModel {
@@ -88,6 +88,10 @@ impl NamespacesViewModel {
         self.model.read(cx).phase().clone()
     }
 
+    pub fn mode(&self, cx: &App) -> WatchMode {
+        self.model.read(cx).snapshot().mode()
+    }
+
     pub fn item_count(&self, cx: &App) -> usize {
         self.model.read(cx).item_count()
     }
diff --git a/src/view_models/resources/network_policies/mod.rs b/src/view_models/resources/network_policies/mod.rs
index 22769c7..e558817 100644
--- a/src/view_models/resources/network_policies/mod.rs
+++ b/src/view_models/resources/network_policies/mod.rs
@@ -13,7 +13,7 @@ use tokio::sync::watch;
 
 use crate::cluster::ClusterState;
 use crate::models::resources::network_policies::NetworkPoliciesModel;
-use crate::models::{Item, NamespaceScope, ResourceKind, SnapshotPhase};
+use crate::models::{Item, NamespaceScope, ResourceKind, SnapshotPhase, WatchMode};
 use crate::repository;
 
 pub struct NetworkPoliciesViewModel {
@@ -92,6 +92,10 @@ impl NetworkPoliciesViewModel {
         self.model.read(cx).phase().clone()
     }
 
+    pub fn mode(&self, cx: &App) -> WatchMode {
+        self.model.read(cx).snapshot().mode()
+    }
+
     pub fn item_count(&self, cx: &App) -> usize {
         self.model.read(cx).item_count()
     }
diff --git a/src/view_models/resources/node/mod.rs b/src/view_models/resources/node/mod.rs
index 742aeba..08dba7a 100644
--- a/src/view_models/resources/node/mod.rs
+++ b/src/view_models/resources/node/mod.rs
@@ -14,7 +14,7 @@ use tokio::sync::watch;
 use crate::cluster::ClusterState;
 use crate::models::resources::node::NodesModel;
 use crate::models::resources::node::detail::{NodeDetailState, UsageHistory};
-use crate::models::{Item, NamespaceScope, ResourceKind, SnapshotPhase};
+use crate::models::{Item, NamespaceScope, ResourceKind, SnapshotPhase, WatchMode};
 use crate::repository;
 
 pub struct NodesViewModel {
@@ -100,6 +100,10 @@ impl NodesViewModel {
         self.model.read(cx).phase().clone()
     }
 
+    pub fn mode(&self, cx: &App) -> WatchMode {
+        self.model.read(cx).snapshot().mode()
+    }
+
     pub fn item_count(&self, cx: &App) -> usize {
         self.model.read(cx).item_count()
     }
diff --git a/src/view_models/resources/persistent_volume_claims/mod.rs b/src/view_models/resources/persistent_volume_claims/mod.rs
index 8e5bd1a..7ce2e32 100644
--- a/src/view_models/resources/persistent_volume_claims/mod.rs
+++ b/src/view_models/resources/persistent_volume_claims/mod.rs
@@ -13,7 +13,7 @@ use tokio::sync::watch;
 
 use crate::cluster::ClusterState;
 use crate::models::resources::persistent_volume_claims::PersistentVolumeClaimsModel;
-use crate::models::{Item, NamespaceScope, ResourceKind, SnapshotPhase};
+use crate::models::{Item, NamespaceScope, ResourceKind, SnapshotPhase, WatchMode};
 use crate::repository;
 
 pub struct PersistentVolumeClaimsViewModel {
@@ -92,6 +92,10 @@ impl PersistentVolumeClaimsViewModel {
         self.model.read(cx).phase().clone()
     }
 
+    pub fn mode(&self, cx: &App) -> WatchMode {
+        self.model.read(cx).snapshot().mode()
+    }
+
     pub fn item_count(&self, cx: &App) -> usize {
         self.model.read(cx).item_count()
     }
diff --git a/src/view_models/resources/persistent_volumes/mod.rs b/src/view_models/resources/persistent_volumes/mod.rs
index 7c18409..008785d 100644
--- a/src/view_models/resources/persistent_volumes/mod.rs
+++ b/src/view_models/resources/persistent_volumes/mod.rs
@@ -13,7 +13,7 @@ use tokio::sync::watch;
 
 use crate::cluster::ClusterState;
 use crate::models::resources::persistent_volumes::PersistentVolumesModel;
-use crate::models::{Item, NamespaceScope, ResourceKind, SnapshotPhase};
+use crate::models::{Item, NamespaceScope, ResourceKind, SnapshotPhase, WatchMode};
 use crate::repository;
 
 pub struct PersistentVolumesViewModel {
@@ -92,6 +92,10 @@ impl PersistentVolumesViewModel {
         self.model.read(cx).phase().clone()
     }
 
+    pub fn mode(&self, cx: &App) -> WatchMode {
+        self.model.read(cx).snapshot().mode()
+    }
+
     pub fn item_count(&self, cx: &App) -> usize {
         self.model.read(cx).item_count()
     }
diff --git a/src/view_models/resources/pod/mod.rs b/src/view_models/resources/pod/mod.rs
index 23785e4..ba070c2 100644
--- a/src/view_models/resources/pod/mod.rs
+++ b/src/view_models/resources/pod/mod.rs
@@ -17,7 +17,7 @@ use tokio::sync::watch;
 use crate::cluster::ClusterState;
 use crate::models::resources::pod::PodsModel;
 use crate::models::resources::pod::detail::{PodDetailState, PodDetailUpdate, UsageHistory};
-use crate::models::{Item, NamespaceScope, ResourceKind, SnapshotPhase};
+use crate::models::{Item, NamespaceScope, ResourceKind, SnapshotPhase, WatchMode};
 use crate::repository;
 
 pub struct PodsViewModel {
@@ -103,6 +103,10 @@ impl PodsViewModel {
         self.model.read(cx).phase().clone()
     }
 
+    pub fn mode(&self, cx: &App) -> WatchMode {
+        self.model.read(cx).snapshot().mode()
+    }
+
     pub fn item_count(&self, cx: &App) -> usize {
         self.model.read(cx).item_count()
     }
diff --git a/src/view_models/resources/pod_disruption_budgets/mod.rs b/src/view_models/resources/pod_disruption_budgets/mod.rs
index 112f01c..cb432a2 100644
--- a/src/view_models/resources/pod_disruption_budgets/mod.rs
+++ b/src/view_models/resources/pod_disruption_budgets/mod.rs
@@ -13,7 +13,7 @@ use tokio::sync::watch;
 
 use crate::cluster::ClusterState;
 use crate::models::resources::pod_disruption_budgets::PodDisruptionBudgetsModel;
-use crate::models::{Item, NamespaceScope, ResourceKind, SnapshotPhase};
+use crate::models::{Item, NamespaceScope, ResourceKind, SnapshotPhase, WatchMode};
 use crate::repository;
 
 pub struct PodDisruptionBudgetsViewModel {
@@ -92,6 +92,10 @@ impl PodDisruptionBudgetsViewModel {
         self.model.read(cx).phase().clone()
     }
 
+    pub fn mode(&self, cx: &App) -> WatchMode {
+        self.model.read(cx).snapshot().mode()
+    }
+
     pub fn item_count(&self, cx: &App) -> usize {
         self.model.read(cx).item_count()
     }
diff --git a/src/view_models/resources/replica_sets/mod.rs b/src/view_models/resources/replica_sets/mod.rs
index d02ebba..d1cbfa8 100644
--- a/src/view_models/resources/replica_sets/mod.rs
+++ b/src/view_models/resources/replica_sets/mod.rs
@@ -13,7 +13,7 @@ use tokio::sync::watch;
 
 use crate::cluster::ClusterState;
 use crate::models::resources::replica_sets::ReplicaSetsModel;
-use crate::models::{Item, NamespaceScope, ResourceKind, SnapshotPhase};
+use crate::models::{Item, NamespaceScope, ResourceKind, SnapshotPhase, WatchMode};
 use crate::repository;
 
 pub struct ReplicaSetsViewModel {
@@ -92,6 +92,10 @@ impl ReplicaSetsViewModel {
         self.model.read(cx).phase().clone()
     }
 
+    pub fn mode(&self, cx: &App) -> WatchMode {
+        self.model.read(cx).snapshot().mode()
+    }
+
     pub fn item_count(&self, cx: &App) -> usize {
         self.model.read(cx).item_count()
     }
diff --git a/src/view_models/resources/replication_controllers/mod.rs b/src/view_models/resources/replication_controllers/mod.rs
index d1080f7..0aeb064 100644
--- a/src/view_models/resources/replication_controllers/mod.rs
+++ b/src/view_models/resources/replication_controllers/mod.rs
@@ -13,7 +13,7 @@ use tokio::sync::watch;
 
 use crate::cluster::ClusterState;
 use crate::models::resources::replication_controllers::ReplicationControllersModel;
-use crate::models::{Item, NamespaceScope, ResourceKind, SnapshotPhase};
+use crate::models::{Item, NamespaceScope, ResourceKind, SnapshotPhase, WatchMode};
 use crate::repository;
 
 pub struct ReplicationControllersViewModel {
@@ -92,6 +92,10 @@ impl ReplicationControllersViewModel {
         self.model.read(cx).phase().clone()
     }
 
+    pub fn mode(&self, cx: &App) -> WatchMode {
+        self.model.read(cx).snapshot().mode()
+    }
+
     pub fn item_count(&self, cx: &App) -> usize {
         self.model.read(cx).item_count()
     }
diff --git a/src/view_models/resources/role_bindings/mod.rs b/src/view_models/resources/role_bindings/mod.rs
index 96f5358..d1b9ea2 100644
--- a/src/view_models/resources/role_bindings/mod.rs
+++ b/src/view_models/resources/role_bindings/mod.rs
@@ -13,7 +13,7 @@ use tokio::sync::watch;
 
 use crate::cluster::ClusterState;
 use crate::models::resources::role_bindings::RoleBindingsModel;
-use crate::models::{Item, NamespaceScope, ResourceKind, SnapshotPhase};
+use crate::models::{Item, NamespaceScope, ResourceKind, SnapshotPhase, WatchMode};
 use crate::repository;
 
 pub struct RoleBindingsViewModel {
@@ -92,6 +92,10 @@ impl RoleBindingsViewModel {
         self.model.read(cx).phase().clone()
     }
 
+    pub fn mode(&self, cx: &App) -> WatchMode {
+        self.model.read(cx).snapshot().mode()
+    }
+
     pub fn item_count(&self, cx: &App) -> usize {
         self.model.read(cx).item_count()
     }
diff --git a/src/view_models/resources/roles/mod.rs b/src/view_models/resources/roles/mod.rs
index bae551c..d3b315b 100644
--- a/src/view_models/resources/roles/mod.rs
+++ b/src/view_models/resources/roles/mod.rs
@@ -13,7 +13,7 @@ use tokio::sync::watch;
 
 use crate::cluster::ClusterState;
 use crate::models::resources::roles::RolesModel;
-use crate::models::{Item, NamespaceScope, ResourceKind, SnapshotPhase};
+use crate::models::{Item, NamespaceScope, ResourceKind, SnapshotPhase, WatchMode};
 use crate::repository;
 
 pub struct RolesViewModel {
@@ -87,6 +87,10 @@ impl RolesViewModel {
         self.model.read(cx).phase().clone()
     }
 
+    pub fn mode(&self, cx: &App) -> WatchMode {
+        self.model.read(cx).snapshot().mode()
+    }
+
     pub fn item_count(&self, cx: &App) -> usize {
         self.model.read(cx).item_count()
     }
diff --git a/src/view_models/resources/secrets/mod.rs b/src/view_models/resources/secrets/mod.rs
index db7d0cb..03f6baa 100644
--- a/src/view_models/resources/secrets/mod.rs
+++ b/src/view_models/resources/secrets/mod.rs
@@ -13,7 +13,7 @@ use tokio::sync::watch;
 
 use crate::cluster::ClusterState;
 use crate::models::resources::secrets::SecretsModel;
-use crate::models::{Item, NamespaceScope, ResourceKind, SnapshotPhase};
+use crate::models::{Item, NamespaceScope, ResourceKind, SnapshotPhase, WatchMode};
 use crate::repository;
 
 pub struct SecretsViewModel {
@@ -87,6 +87,10 @@ impl SecretsViewModel {
         self.model.read(cx).phase().clone()
     }
 
+    pub fn mode(&self, cx: &App) -> WatchMode {
+        self.model.read(cx).snapshot().mode()
+    }
+
     pub fn item_count(&self, cx: &App) -> usize {
         self.model.read(cx).item_count()
     }
diff --git a/src/view_models/resources/service_accounts/mod.rs b/src/view_models/resources/service_accounts/mod.rs
index 2aec2bb..e8e99ed 100644
--- a/src/view_models/resources/service_accounts/mod.rs
+++ b/src/view_models/resources/service_accounts/mod.rs
@@ -13,7 +13,7 @@ use tokio::sync::watch;
 
 use crate::cluster::ClusterState;
 use crate::models::resources::service_accounts::ServiceAccountsModel;
-use crate::models::{Item, NamespaceScope, ResourceKind, SnapshotPhase};
+use crate::models::{Item, NamespaceScope, ResourceKind, SnapshotPhase, WatchMode};
 use crate::repository;
 
 pub struct ServiceAccountsViewModel {
@@ -92,6 +92,10 @@ impl ServiceAccountsViewModel {
         self.model.read(cx).phase().clone()
     }
 
+    pub fn mode(&self, cx: &App) -> WatchMode {
+        self.model.read(cx).snapshot().mode()
+    }
+
     pub fn item_count(&self, cx: &App) -> usize {
         self.model.read(cx).item_count()
     }
diff --git a/src/view_models/resources/services/mod.rs b/src/view_models/resources/services/mod.rs
index ee7cba2..6f2e518 100644
--- a/src/view_models/resources/services/mod.rs
+++ b/src/view_models/resources/services/mod.rs
@@ -13,7 +13,7 @@ use tokio::sync::watch;
 
 use crate::cluster::ClusterState;
 use crate::models::resources::services::ServicesModel;
-use crate::models::{Item, NamespaceScope, ResourceKind, SnapshotPhase};
+use crate::models::{Item, NamespaceScope, ResourceKind, SnapshotPhase, WatchMode};
 use crate::repository;
 
 pub struct ServicesViewModel {
@@ -87,6 +87,10 @@ impl ServicesViewModel {
         self.model.read(cx).phase().clone()
     }
 
+    pub fn mode(&self, cx: &App) -> WatchMode {
+        self.model.read(cx).snapshot().mode()
+    }
+
     pub fn item_count(&self, cx: &App) -> usize {
         self.model.read(cx).item_count()
     }
diff --git a/src/view_models/resources/stateful_sets/mod.rs b/src/view_models/resources/stateful_sets/mod.rs
index 7230cc4..4d6b653 100644
--- a/src/view_models/resources/stateful_sets/mod.rs
+++ b/src/view_models/resources/stateful_sets/mod.rs
@@ -13,7 +13,7 @@ use tokio::sync::watch;
 
 use crate::cluster::ClusterState;
 use crate::models::resources::stateful_sets::StatefulSetsModel;
-use crate::models::{Item, NamespaceScope, ResourceKind, SnapshotPhase};
+use crate::models::{Item, NamespaceScope, ResourceKind, SnapshotPhase, WatchMode};
 use crate::repository;
 
 pub struct StatefulSetsViewModel {
@@ -92,6 +92,10 @@ impl StatefulSetsViewModel {
         self.model.read(cx).phase().clone()
     }
 
+    pub fn mode(&self, cx: &App) -> WatchMode {
+        self.model.read(cx).snapshot().mode()
+    }
+
     pub fn item_count(&self, cx: &App) -> usize {
         self.model.read(cx).item_count()
     }
diff --git a/src/view_models/resources/storage_classes/mod.rs b/src/view_models/resources/storage_classes/mod.rs
index 8cb9be4..57781e3 100644
--- a/src/view_models/resources/storage_classes/mod.rs
+++ b/src/view_models/resources/storage_classes/mod.rs
@@ -13,7 +13,7 @@ use tokio::sync::watch;
 
 use crate::cluster::ClusterState;
 use crate::models::resources::storage_classes::StorageClassesModel;
-use crate::models::{Item, NamespaceScope, ResourceKind, SnapshotPhase};
+use crate::models::{Item, NamespaceScope, ResourceKind, SnapshotPhase, WatchMode};
 use crate::repository;
 
 pub struct StorageClassesViewModel {
@@ -92,6 +92,10 @@ impl StorageClassesViewModel {
         self.model.read(cx).phase().clone()
     }
 
+    pub fn mode(&self, cx: &App) -> WatchMode {
+        self.model.read(cx).snapshot().mode()
+    }
+
     pub fn item_count(&self, cx: &App) -> usize {
         self.model.read(cx).item_count()
     }