diff --git a/Cargo.lock b/Cargo.lock
index 4c4911cd..2ab2aa3a 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -1451,6 +1451,8 @@ dependencies = [
  "http 1.4.0",
  "http-body 1.0.1",
  "http-body-util",
+ "hyper 1.8.1",
+ "hyper-util",
  "itoa",
  "matchit",
  "memchr",
@@ -1459,6 +1461,7 @@ dependencies = [
  "pin-project-lite",
  "serde_core",
  "sync_wrapper 1.0.2",
+ "tokio",
  "tower",
  "tower-layer",
  "tower-service",
@@ -2481,6 +2484,7 @@ dependencies = [
  "anyhow",
  "async-signal",
  "async-trait",
+ "axum",
  "custom_debug",
  "dipper-core",
  "dipper-iisa",
diff --git a/bin/dipper-service/Cargo.toml b/bin/dipper-service/Cargo.toml
index 67375c3e..94e2c1e6 100644
--- a/bin/dipper-service/Cargo.toml
+++ b/bin/dipper-service/Cargo.toml
@@ -7,6 +7,7 @@ edition = "2024"
 anyhow.workspace = true
 async-signal = "0.2.10"
 async-trait.workspace = true
+axum = { version = "0.8", default-features = false, features = ["http1", "tokio"] }
 custom_debug = "0.6.1"
 dipper-core = { version = "0.1.0", path = "../../dipper-core" }
 dipper-iisa = { path = "../../dipper-iisa" }
diff --git a/bin/dipper-service/src/chain_client.rs b/bin/dipper-service/src/chain_client.rs
index bf39bcd1..35ad6d90 100644
--- a/bin/dipper-service/src/chain_client.rs
+++ b/bin/dipper-service/src/chain_client.rs
@@ -169,3 +169,85 @@ impl<T: ChainClient + Send + Sync + ?Sized> ChainClient for Arc<T> {
         (**self).latest_block_timestamp().await
     }
 }
+
+/// Runs the periodic RCA EIP-712 domain refresh until `stop_rx` fires.
+///
+/// Generic over the refresh action so the stop wiring is unit testable without
+/// a live chain. The first (immediate) interval tick is skipped; thereafter
+/// each tick invokes `refresh`, whose errors are logged and swallowed (the
+/// current domain is kept). Returns `Ok(())` on stop.
+pub async fn run_domain_refresh<F, Fut>(
+    interval: std::time::Duration,
+    mut stop_rx: tokio::sync::mpsc::Receiver<()>,
+    mut refresh: F,
+) -> anyhow::Result<()>
+where
+    F: FnMut() -> Fut,
+    Fut: std::future::Future<Output = Result<bool, ChainClientError>>,
+{
+    let mut ticker = tokio::time::interval(interval);
+    ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
+    ticker.tick().await; // the first tick fires immediately; skip it
+    loop {
+        tokio::select! { biased;
+            _ = stop_rx.recv() => return Ok(()),
+            _ = ticker.tick() => {
+                if let Err(err) = refresh().await {
+                    tracing::warn!(
+                        error = %err,
+                        "RCA EIP-712 domain refresh failed; keeping the current domain"
+                    );
+                }
+            }
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use std::{
+        sync::{
+            Arc,
+            atomic::{AtomicUsize, Ordering},
+        },
+        time::Duration,
+    };
+
+    use super::run_domain_refresh;
+
+    /// The refresh loop must exit promptly when stopped, even mid-wait, so it
+    /// participates in graceful shutdown instead of being a detached task.
+    #[tokio::test]
+    async fn refresh_loop_stops_on_signal() {
+        let (tx, rx) = tokio::sync::mpsc::channel(1);
+        let calls = Arc::new(AtomicUsize::new(0));
+        let calls_in = calls.clone();
+
+        // A long interval so no refresh tick fires during the test; the stop
+        // arm is what must end the loop.
+        let handle = tokio::spawn(run_domain_refresh(
+            Duration::from_secs(3600),
+            rx,
+            move || {
+                let calls = calls_in.clone();
+                async move {
+                    calls.fetch_add(1, Ordering::SeqCst);
+                    Ok(true)
+                }
+            },
+        ));
+
+        tx.send(()).await.unwrap();
+        let result = tokio::time::timeout(Duration::from_secs(5), handle)
+            .await
+            .expect("refresh loop did not stop on signal")
+            .expect("refresh task panicked");
+
+        assert!(result.is_ok());
+        assert_eq!(
+            calls.load(Ordering::SeqCst),
+            0,
+            "no refresh should have fired before the stop signal"
+        );
+    }
+}
diff --git a/bin/dipper-service/src/config.rs b/bin/dipper-service/src/config.rs
index 460555b7..138c4aee 100644
--- a/bin/dipper-service/src/config.rs
+++ b/bin/dipper-service/src/config.rs
@@ -90,6 +90,30 @@ pub struct Config {
     /// the pool with the registry and background services; size accordingly.
     #[serde(default = "default_worker_concurrency")]
     pub worker_concurrency: usize,
+    /// The health endpoint configuration. When unset, no health server is
+    /// started.
+    #[serde(default)]
+    pub health: Option<HealthConfig>,
+}
+
+/// Configuration for the HTTP health endpoint used by orchestrator liveness
+/// probes.
+#[serde_as]
+#[derive(Debug, serde::Deserialize)]
+pub struct HealthConfig {
+    /// The health server listen address (e.g. `0.0.0.0:8080`).
+    #[serde_as(as = "serde_with::DisplayFromStr")]
+    pub listen_addr: std::net::SocketAddr,
+
+    /// Staleness threshold in seconds after which the worker is reported
+    /// unhealthy. Defaults to [`crate::health::DEFAULT_HEALTH_THRESHOLD`].
+    #[serde(default = "default_health_threshold")]
+    #[serde_as(as = "serde_with::DurationSeconds")]
+    pub threshold: Duration,
+}
+
+fn default_health_threshold() -> Duration {
+    crate::health::DEFAULT_HEALTH_THRESHOLD
 }
 
 /// The IISA (Indexing Indexer Selection Algorithm) service configuration
diff --git a/bin/dipper-service/src/health.rs b/bin/dipper-service/src/health.rs
new file mode 100644
index 00000000..d6ef865c
--- /dev/null
+++ b/bin/dipper-service/src/health.rs
@@ -0,0 +1,288 @@
+//! Worker liveness watermark and a minimal HTTP health endpoint.
+//!
+//! The supervisor ([`crate::supervisor`]) catches a worker that *exits*
+//! (panic or error return). It cannot catch a worker that is *wedged* — alive
+//! but making no progress (e.g. parked on an await that the per-call timeouts
+//! somehow didn't bound). This module closes that gap: the worker ticks a
+//! progress watermark every loop iteration, and a small health server reports
+//! 503 once the watermark goes stale so an external orchestrator (k8s liveness
+//! probe) can restart the wedged process.
+
+use std::{
+    net::SocketAddr,
+    sync::{
+        Arc,
+        atomic::{AtomicI64, Ordering},
+    },
+    time::Duration,
+};
+
+use axum::{Router, extract::State, http::StatusCode, response::IntoResponse, routing::get};
+use time::OffsetDateTime;
+use tokio::{net::TcpListener, sync::mpsc};
+
+/// Default staleness threshold after which the worker is considered wedged.
+///
+/// Must exceed the worker's worst-case time between progress ticks: the poll
+/// period plus a single job's backstop timeout
+/// ([`crate::worker::service`]'s `PROCESS_JOB_TIMEOUT`, 300s). 600s leaves
+/// comfortable headroom so a legitimately slow job never trips the probe.
+pub const DEFAULT_HEALTH_THRESHOLD: Duration = Duration::from_secs(600);
+
+/// Shared liveness watermark: the unix-seconds timestamp at which the worker
+/// last made progress. The worker ticks it via [`Liveness::record_progress`];
+/// the health server reads it via [`Liveness::is_healthy`].
+#[derive(Clone)]
+pub struct Liveness {
+    last_progress: Arc<AtomicI64>,
+}
+
+fn now_unix() -> i64 {
+    OffsetDateTime::now_utc().unix_timestamp()
+}
+
+impl Liveness {
+    /// Creates a watermark seeded to now, so a freshly started worker is
+    /// considered live (startup grace).
+    pub fn new() -> Self {
+        Self {
+            last_progress: Arc::new(AtomicI64::new(now_unix())),
+        }
+    }
+
+    /// Records that the worker just made progress.
+    pub fn record_progress(&self) {
+        self.last_progress.store(now_unix(), Ordering::Relaxed);
+    }
+
+    /// Whether the worker made progress within `threshold` of `now_unix`.
+    ///
+    /// Pure in its inputs so it is unit testable without touching the clock.
+    pub fn is_healthy_at(&self, now_unix: i64, threshold: Duration) -> bool {
+        let age = now_unix.saturating_sub(self.last_progress.load(Ordering::Relaxed));
+        age <= threshold.as_secs() as i64
+    }
+
+    /// [`Liveness::is_healthy_at`] against the current clock.
+    pub fn is_healthy(&self, threshold: Duration) -> bool {
+        self.is_healthy_at(now_unix(), threshold)
+    }
+}
+
+impl Default for Liveness {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+/// Handle to stop the health server.
+pub struct Handle {
+    tx_stop: mpsc::Sender<()>,
+}
+
+impl Handle {
+    /// Stops the health server.
+    pub async fn stop(self) {
+        if self.tx_stop.is_closed() {
+            return;
+        }
+        let _ = self.tx_stop.send(()).await;
+        self.tx_stop.closed().await;
+    }
+}
+
+/// Binds the health server and returns a stop handle plus its run future.
+///
+/// The listener is bound eagerly so a bind failure surfaces at startup (and so
+/// callers/tests can read the actual bound address).
+pub async fn new(
+    addr: SocketAddr,
+    liveness: Liveness,
+    threshold: Duration,
+) -> anyhow::Result<(
+    Handle,
+    SocketAddr,
+    impl std::future::Future<Output = anyhow::Result<()>>,
+)> {
+    let listener = TcpListener::bind(addr).await?;
+    let local_addr = listener.local_addr()?;
+    let (tx_stop, rx_stop) = mpsc::channel(1);
+    let fut = serve(listener, liveness, threshold, rx_stop);
+    Ok((Handle { tx_stop }, local_addr, fut))
+}
+
+/// State shared with the [`health`] handler.
+#[derive(Clone)]
+struct HealthState {
+    liveness: Liveness,
+    threshold: Duration,
+}
+
+/// `GET /health`: 200 while the worker watermark is fresh, 503 once it has gone
+/// stale (the worker is wedged and the orchestrator should restart it).
+async fn health(State(state): State<HealthState>) -> impl IntoResponse {
+    if state.liveness.is_healthy(state.threshold) {
+        (StatusCode::OK, "ok")
+    } else {
+        (StatusCode::SERVICE_UNAVAILABLE, "worker stalled")
+    }
+}
+
+/// Serves health requests until `stop_rx` fires.
+///
+/// Backed by axum/hyper (already in the dependency tree via the RPC servers),
+/// so request parsing, per-connection isolation, connection timeouts and
+/// graceful shutdown are handled by the HTTP stack rather than by hand.
+async fn serve(
+    listener: TcpListener,
+    liveness: Liveness,
+    threshold: Duration,
+    mut stop_rx: mpsc::Receiver<()>,
+) -> anyhow::Result<()> {
+    tracing::info!(addr = %listener.local_addr()?, "health server listening");
+    let app = Router::new()
+        .route("/health", get(health))
+        .with_state(HealthState {
+            liveness,
+            threshold,
+        });
+    axum::serve(listener, app)
+        .with_graceful_shutdown(async move {
+            let _ = stop_rx.recv().await;
+        })
+        .await?;
+    Ok(())
+}
+
+#[cfg(test)]
+mod tests {
+    use std::time::Duration;
+
+    use tokio::{
+        io::{AsyncReadExt, AsyncWriteExt},
+        net::TcpStream,
+    };
+
+    use super::*;
+
+    #[test]
+    fn fresh_watermark_is_healthy() {
+        let liveness = Liveness::new();
+        let now = now_unix();
+        assert!(liveness.is_healthy_at(now, Duration::from_secs(600)));
+    }
+
+    #[test]
+    fn stale_watermark_is_unhealthy() {
+        let liveness = Liveness::new();
+        // Pretend "now" is well past the threshold since the watermark was set.
+        let now = now_unix() + 10_000;
+        assert!(
+            !liveness.is_healthy_at(now, Duration::from_secs(600)),
+            "a watermark older than the threshold must report unhealthy"
+        );
+    }
+
+    #[test]
+    fn boundary_is_inclusive_healthy() {
+        let liveness = Liveness::new();
+        let base = now_unix();
+        // Exactly at the threshold is still healthy; one second past is not.
+        assert!(liveness.is_healthy_at(base + 600, Duration::from_secs(600)));
+        assert!(!liveness.is_healthy_at(base + 601, Duration::from_secs(600)));
+    }
+
+    /// Reads an HTTP response's status line from a fresh connection to the
+    /// server.
+    async fn probe(addr: SocketAddr) -> String {
+        let mut stream = TcpStream::connect(addr).await.unwrap();
+        // `Connection: close` so the server closes the socket after responding
+        // and `read_to_end` returns rather than blocking on HTTP/1.1 keep-alive.
+        stream
+            .write_all(b"GET /health HTTP/1.1\r\nHost: x\r\nConnection: close\r\n\r\n")
+            .await
+            .unwrap();
+        let mut buf = Vec::new();
+        stream.read_to_end(&mut buf).await.unwrap();
+        let text = String::from_utf8_lossy(&buf);
+        text.lines().next().unwrap_or_default().to_string()
+    }
+
+    #[tokio::test]
+    async fn server_reports_503_when_stalled() {
+        let liveness = Liveness::new();
+        // A zero threshold makes the (initially fresh) watermark immediately
+        // stale, so we exercise the unhealthy branch deterministically.
+        let (handle, addr, fut) = new("127.0.0.1:0".parse().unwrap(), liveness, Duration::ZERO)
+            .await
+            .unwrap();
+        // Ensure the watermark is at least a second old so age > 0.
+        let server = tokio::spawn(fut);
+
+        // Wait a moment so the seeded watermark is strictly in the past.
+        tokio::time::sleep(Duration::from_millis(1100)).await;
+        let status = probe(addr).await;
+        assert!(
+            status.contains("503"),
+            "expected 503 when stalled, got: {status:?}"
+        );
+
+        handle.stop().await;
+        let _ = server.await;
+    }
+
+    #[tokio::test]
+    async fn server_reports_200_when_live() {
+        let liveness = Liveness::new();
+        let (handle, addr, fut) = new(
+            "127.0.0.1:0".parse().unwrap(),
+            liveness,
+            Duration::from_secs(600),
+        )
+        .await
+        .unwrap();
+        let server = tokio::spawn(fut);
+
+        let status = probe(addr).await;
+        assert!(
+            status.contains("200"),
+            "expected 200 when live, got: {status:?}"
+        );
+
+        handle.stop().await;
+        let _ = server.await;
+    }
+
+    /// A client that connects but never sends must not wedge the server: other
+    /// probes are still answered promptly, and graceful shutdown still
+    /// completes. hyper isolates each connection, but this guards against a
+    /// future regression that serves connections without that isolation.
+    #[tokio::test]
+    async fn stalled_client_does_not_block_other_probes() {
+        let liveness = Liveness::new();
+        let (handle, addr, fut) = new(
+            "127.0.0.1:0".parse().unwrap(),
+            liveness,
+            Duration::from_secs(600),
+        )
+        .await
+        .unwrap();
+        let server = tokio::spawn(fut);
+
+        // Open a connection and never write to it; its server-side read blocks.
+        let _stalled = TcpStream::connect(addr).await.unwrap();
+
+        // A well-behaved probe must still get a response without waiting on the
+        // stalled client (well under CONN_TIMEOUT).
+        let status = tokio::time::timeout(Duration::from_secs(2), probe(addr))
+            .await
+            .expect("a stalled client blocked an independent probe");
+        assert!(status.contains("200"), "expected 200, got: {status:?}");
+
+        // Shutdown must not wait on the stalled connection either.
+        tokio::time::timeout(Duration::from_secs(2), handle.stop())
+            .await
+            .expect("a stalled client blocked shutdown");
+        let _ = server.await;
+    }
+}
diff --git a/bin/dipper-service/src/main.rs b/bin/dipper-service/src/main.rs
index 0b99f510..3f40c46d 100644
--- a/bin/dipper-service/src/main.rs
+++ b/bin/dipper-service/src/main.rs
@@ -21,10 +21,12 @@ mod cancel_dispatch;
 mod chain_client;
 mod config;
 mod db;
+mod health;
 mod indexer_rpc_client;
 mod network;
 mod registry;
 mod signing;
+mod supervisor;
 #[cfg(test)]
 mod test_support;
 mod worker;
@@ -124,31 +126,32 @@ pub async fn main() -> anyhow::Result<()> {
 
     // Background refresh so a running dipper follows an in-place contract upgrade
     // without a restart. Refresh failures keep the current domain and only warn.
-    if let Some(cfg) = conf.chain_client.as_ref().filter(|cfg| cfg.enabled) {
-        let cfg = cfg.clone();
-        let rca_domain = Arc::clone(&rca_domain);
-        tokio::spawn(async move {
-            let mut ticker = tokio::time::interval(cfg.domain_refresh_interval);
-            ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
-            ticker.tick().await; // the first tick fires immediately; skip it
-            loop {
-                ticker.tick().await;
-                if let Err(err) = chain_client::refresh_rca_eip712_domain(
-                    &cfg,
-                    chain_id,
-                    recurring_collector,
-                    &rca_domain,
-                )
-                .await
-                {
-                    tracing::warn!(
-                        error = %err,
-                        "RCA EIP-712 domain refresh failed; keeping the current domain"
-                    );
+    // Built here but spawned into the supervised task tree below, with a stop
+    // handle, rather than detached — so it shares the same shutdown and
+    // supervision as every other long-running task.
+    let domain_refresh_handle =
+        if let Some(cfg) = conf.chain_client.as_ref().filter(|cfg| cfg.enabled) {
+            let cfg = cfg.clone();
+            let rca_domain = Arc::clone(&rca_domain);
+            let interval = cfg.domain_refresh_interval;
+            let (tx_stop, rx_stop) = tokio::sync::mpsc::channel(1);
+            let service = chain_client::run_domain_refresh(interval, rx_stop, move || {
+                let cfg = cfg.clone();
+                let rca_domain = Arc::clone(&rca_domain);
+                async move {
+                    chain_client::refresh_rca_eip712_domain(
+                        &cfg,
+                        chain_id,
+                        recurring_collector,
+                        &rca_domain,
+                    )
+                    .await
                 }
-            }
-        });
-    }
+            });
+            Some((tx_stop, service))
+        } else {
+            None
+        };
 
     // Initialize the different components
 
@@ -373,6 +376,9 @@ pub async fn main() -> anyhow::Result<()> {
     // dispatched so it switches from 300s idle polling to 5s immediately.
     let chain_listener_notify = Arc::new(tokio::sync::Notify::new());
 
+    //- Worker liveness watermark, shared with the health endpoint below.
+    let worker_liveness = health::Liveness::new();
+
     //- The worker service
     let (worker_handle, worker_service) = {
         // Each loop can hold up to three pooled connections at once and shares
@@ -435,6 +441,7 @@ pub async fn main() -> anyhow::Result<()> {
                 .map(|c| c.bypass_chain_clock_defenses)
                 .unwrap_or(false),
             chain_listener_chain_id: conf.chain_listener.as_ref().map(|c| c.chain_id),
+            liveness: worker_liveness.clone(),
             reassess_lock,
             unresponsive_breaker,
             dips_accepting_cache,
@@ -571,15 +578,56 @@ pub async fn main() -> anyhow::Result<()> {
     };
     tracing::info!("initialized Admin RPC service");
 
+    //- The health endpoint (optional). Reports 503 once the worker watermark
+    //  goes stale so an orchestrator can restart a wedged process.
+    let health_handle = if let Some(health_conf) = conf.health.as_ref() {
+        // The worker only ticks the watermark once per loop iteration, so the
+        // worst-case gap between ticks is one job's backstop timeout. A
+        // threshold at or below that would trip the probe during a legitimately
+        // slow job and cause restart loops; warn rather than fail so a
+        // deliberate aggressive setting is still possible.
+        if health_conf.threshold <= worker::service::PROCESS_JOB_TIMEOUT {
+            tracing::warn!(
+                threshold_secs = health_conf.threshold.as_secs(),
+                process_job_timeout_secs = worker::service::PROCESS_JOB_TIMEOUT.as_secs(),
+                "health threshold is at or below the worker's per-job backstop timeout; a slow \
+                 job may trip the liveness probe and cause spurious restarts"
+            );
+        }
+        let (handle, addr, service) = health::new(
+            health_conf.listen_addr,
+            worker_liveness.clone(),
+            health_conf.threshold,
+        )
+        .await?;
+        tracing::info!(%addr, "initialized health endpoint");
+        Some((handle, service))
+    } else {
+        None
+    };
+
     // Construct the task tree
     let mut task_tree = JoinSet::new();
 
+    // Shared shutdown coordination. A signal or an unexpected task exit both
+    // drive the same graceful stop sequence below.
+    let shutdown = supervisor::Shutdown::new();
+
     let indexer_urls_task_handle = task_tree.spawn(indexer_urls_service);
     tracing::debug!(task_id=%indexer_urls_task_handle.id(), "Indexer URLs service started");
 
     let worker_task_handle = task_tree.spawn(worker_service);
     tracing::debug!(task_id=%worker_task_handle.id(), "Worker service started");
 
+    // Spawn the RCA domain refresh if the chain client is enabled
+    let domain_refresh_stop_handle = if let Some((tx_stop, service)) = domain_refresh_handle {
+        let task_handle = task_tree.spawn(service);
+        tracing::debug!(task_id=%task_handle.id(), "RCA domain refresh started");
+        Some(tx_stop)
+    } else {
+        None
+    };
+
     // Spawn the reassignment service if enabled
     let reassignment_stop_handle = if let Some((handle, service)) = reassignment_handle {
         let task_handle = task_tree.spawn(service);
@@ -625,22 +673,50 @@ pub async fn main() -> anyhow::Result<()> {
         None
     };
 
+    // Spawn the health endpoint if enabled
+    let health_stop_handle = if let Some((handle, service)) = health_handle {
+        let task_handle = task_tree.spawn(service);
+        tracing::debug!(task_id=%task_handle.id(), "Health endpoint started");
+        Some(handle)
+    } else {
+        None
+    };
+
     let admin_rpc_task_handle = task_tree.spawn(admin_rpc_service);
     tracing::debug!(task_id=%admin_rpc_task_handle.id(), "Admin RPC service started");
 
+    let shutdown_coordinator = shutdown.clone();
     let signal_handler_task_handle = task_tree.spawn(async move {
-        let signal = signal_task().await;
-        match signal {
-            Ok(AppSignal::Shutdown) => {
-                tracing::info!("shutting down");
-            }
-            Err(err) => {
-                tracing::error!(error=?err, "signal handler registration failed. shutting down");
+        // Wake on either an OS signal or an unexpected critical-task exit
+        // (the supervisor requests shutdown in the latter case).
+        tokio::select! {
+            signal = signal_task() => match signal {
+                Ok(AppSignal::Shutdown) => {
+                    tracing::info!("shutting down");
+                }
+                Err(err) => {
+                    tracing::error!(error=?err, "signal handler registration failed. shutting down");
+                }
+            },
+            _ = shutdown_coordinator.requested_signal() => {
+                tracing::warn!("shutting down due to an unexpected critical task exit");
             }
         }
 
+        // Ensure the flag is set on the signal path too, so the supervisor
+        // treats the resulting service completions as a deliberate shutdown.
+        shutdown_coordinator.request();
+
         // Stop all services in reverse dependency order, so a service is stopped before the
         // services it depends on.
+
+        // Stop the health endpoint first; nothing depends on it.
+        if let Some(handle) = health_stop_handle {
+            tracing::trace!("stopping Health endpoint");
+            handle.stop().await;
+            tracing::trace!("stopped Health endpoint");
+        }
+
         tracing::trace!("stopping Admin RPC service");
         admin_rpc_handle.stop().await;
         tracing::trace!("stopped Admin RPC service");
@@ -687,6 +763,13 @@ pub async fn main() -> anyhow::Result<()> {
             tracing::trace!("stopped Entity count cache service");
         }
 
+        // Stop the RCA domain refresh (a background helper to the chain client)
+        if let Some(tx_stop) = domain_refresh_stop_handle {
+            tracing::trace!("stopping RCA domain refresh");
+            let _ = tx_stop.send(()).await;
+            tracing::trace!("stopped RCA domain refresh");
+        }
+
         tracing::trace!("stopping Worker service");
         worker_handle.stop().await;
         tracing::trace!("stopped Worker service");
@@ -709,22 +792,11 @@ pub async fn main() -> anyhow::Result<()> {
     });
     tracing::debug!(task_id=%signal_handler_task_handle.id(), "signal handler registered");
 
-    // Block on the task tree. Wait for all tasks to complete
+    // Supervise the task tree. An unexpected task exit (one that happens before
+    // shutdown was requested) tears the rest of the tree down and returns an
+    // error so the process exits non-zero for the orchestrator to restart.
     tracing::info!("starting service");
-    while let Some(res) = task_tree.join_next_with_id().await {
-        match res {
-            Ok((id, Ok(()))) => {
-                tracing::debug!(task_id=%id, "task completed");
-            }
-            Ok((id, Err(err))) => {
-                tracing::error!(task_id=%id, error=?err, "task failed");
-            }
-            Err(err) => {
-                tracing::error!(task_id=%err.id(), error=?err, "task join error");
-            }
-        }
-    }
-    Ok(())
+    supervisor::supervise(task_tree, &shutdown).await
 }
 
 /// Signals that the application can receive
diff --git a/bin/dipper-service/src/supervisor.rs b/bin/dipper-service/src/supervisor.rs
new file mode 100644
index 00000000..ae205c05
--- /dev/null
+++ b/bin/dipper-service/src/supervisor.rs
@@ -0,0 +1,228 @@
+//! Process supervision for the task tree.
+//!
+//! Every long-running service is spawned into a single [`JoinSet`]. In normal
+//! operation none of them ever completes on its own — they run until the
+//! shutdown sequence stops them. So a task that finishes *before* shutdown was
+//! deliberately requested is, by definition, an unexpected critical-task exit
+//! (a panic, or a loop returning `Err` for an unrecoverable reason).
+//!
+//! Leaving the process running in that state — other services up, the daemon
+//! looking healthy, but a critical task dead — is exactly the silent-stall
+//! failure mode we refuse to allow. [`supervise`] therefore treats any such
+//! exit as fatal: it requests shutdown so the rest of the tree is torn down
+//! cleanly, then returns an error so the process exits non-zero and the
+//! orchestrator restarts it.
+
+use std::sync::{
+    Arc,
+    atomic::{AtomicBool, Ordering},
+};
+
+use tokio::{sync::Notify, task::JoinSet};
+
+/// Shared shutdown coordination between [`supervise`] and the task that runs
+/// the graceful stop sequence.
+///
+/// Cloneable; all clones observe the same state.
+#[derive(Clone, Default)]
+pub struct Shutdown {
+    requested: Arc<AtomicBool>,
+    trigger: Arc<Notify>,
+}
+
+impl Shutdown {
+    pub fn new() -> Self {
+        Self::default()
+    }
+
+    /// Marks shutdown as requested and wakes whoever awaits
+    /// [`Shutdown::requested_signal`]. Idempotent.
+    pub fn request(&self) {
+        self.requested.store(true, Ordering::SeqCst);
+        // `notify_waiters`, not `notify_one`: multiple tasks may be parked in
+        // `requested_signal`, and shutdown must wake all of them. It stores no
+        // permit for future waiters, but none is needed — a waiter that arrives
+        // after this sees the flag via the register-then-check in
+        // `requested_signal` and returns without parking.
+        self.trigger.notify_waiters();
+    }
+
+    /// Whether shutdown has been requested (by a signal or by an unexpected
+    /// task exit).
+    pub fn is_requested(&self) -> bool {
+        self.requested.load(Ordering::SeqCst)
+    }
+
+    /// Resolves once shutdown has been requested.
+    ///
+    /// Uses the register-then-check ordering so a [`Shutdown::request`] racing
+    /// between the flag read and the wait can't be lost.
+    pub async fn requested_signal(&self) {
+        let notified = self.trigger.notified();
+        tokio::pin!(notified);
+        // Register interest before re-reading the flag.
+        notified.as_mut().enable();
+        if self.is_requested() {
+            return;
+        }
+        notified.await;
+    }
+}
+
+/// Drains `task_tree`, treating any task that finishes before shutdown was
+/// requested as a fatal unexpected exit.
+///
+/// On the first such exit it logs, requests shutdown (so the stop-sequence task
+/// tears the rest of the tree down), and keeps draining. Returns `Err` if any
+/// unexpected exit occurred, otherwise `Ok(())` (a deliberate shutdown).
+pub async fn supervise(
+    mut task_tree: JoinSet<anyhow::Result<()>>,
+    shutdown: &Shutdown,
+) -> anyhow::Result<()> {
+    let mut unexpected_exit = false;
+
+    while let Some(res) = task_tree.join_next_with_id().await {
+        match res {
+            Ok((id, Ok(()))) => {
+                tracing::debug!(task_id = %id, "task completed");
+            }
+            Ok((id, Err(err))) => {
+                tracing::error!(task_id = %id, error = ?err, "task failed");
+            }
+            Err(err) => {
+                tracing::error!(task_id = %err.id(), error = ?err, "task join error");
+            }
+        }
+
+        if !shutdown.is_requested() {
+            tracing::error!(
+                "a critical task exited unexpectedly; initiating shutdown so the process \
+                 restarts rather than running on with a dead task"
+            );
+            unexpected_exit = true;
+            shutdown.request();
+        }
+    }
+
+    if unexpected_exit {
+        anyhow::bail!("a critical task exited unexpectedly");
+    }
+    Ok(())
+}
+
+#[cfg(test)]
+mod tests {
+    use std::time::Duration;
+
+    use tokio::task::JoinSet;
+
+    use super::{Shutdown, supervise};
+
+    /// Bound on how long `supervise` may run in a test. A broken impl that
+    /// never requests shutdown would otherwise hang the coordinator stand-in
+    /// forever; this turns that into a deterministic failure.
+    const SUPERVISE_TIMEOUT: Duration = Duration::from_secs(5);
+
+    /// A task finishing before shutdown was requested is fatal: `supervise`
+    /// returns an error and requests shutdown so the rest of the tree is torn
+    /// down. Regression test for the join loop that merely logged a dead task
+    /// and let the process keep running.
+    #[tokio::test]
+    async fn unexpected_task_exit_is_fatal_and_triggers_shutdown() {
+        let shutdown = Shutdown::new();
+        let mut task_tree: JoinSet<anyhow::Result<()>> = JoinSet::new();
+
+        // Stand-in for the stop-sequence task: drains once shutdown is asked
+        // for. If `supervise` fails to request shutdown, this never completes
+        // and the test hangs (caught by the harness timeout).
+        let coordinator = shutdown.clone();
+        task_tree.spawn(async move {
+            coordinator.requested_signal().await;
+            Ok(())
+        });
+
+        // A critical task that exits on its own, with nobody having asked for
+        // shutdown.
+        task_tree.spawn(async { Ok(()) });
+
+        let result = tokio::time::timeout(SUPERVISE_TIMEOUT, supervise(task_tree, &shutdown))
+            .await
+            .expect("supervise did not request shutdown; coordinator never drained");
+
+        assert!(result.is_err(), "an unexpected task exit must be fatal");
+        assert!(
+            shutdown.is_requested(),
+            "an unexpected task exit must request shutdown"
+        );
+    }
+
+    /// When shutdown was deliberately requested, tasks completing afterwards is
+    /// the normal path and must not be reported as fatal.
+    #[tokio::test]
+    async fn deliberate_shutdown_is_not_fatal() {
+        let shutdown = Shutdown::new();
+        let mut task_tree: JoinSet<anyhow::Result<()>> = JoinSet::new();
+
+        // The "signal handler": requests shutdown, then completes.
+        let coordinator = shutdown.clone();
+        task_tree.spawn(async move {
+            coordinator.request();
+            Ok(())
+        });
+
+        // Other services that exit only after shutdown is requested.
+        for _ in 0..3 {
+            let s = shutdown.clone();
+            task_tree.spawn(async move {
+                s.requested_signal().await;
+                Ok(())
+            });
+        }
+
+        let result = tokio::time::timeout(SUPERVISE_TIMEOUT, supervise(task_tree, &shutdown))
+            .await
+            .expect("supervise did not drain after a deliberate shutdown");
+
+        assert!(
+            result.is_ok(),
+            "a deliberate shutdown must not be reported as fatal: {result:?}"
+        );
+    }
+
+    /// A single `request()` must wake *every* parked waiter, not just one.
+    ///
+    /// Both waiters are polled to their suspended (registered) state while the
+    /// flag is still false, then `request()` is called once. Regression test
+    /// for `notify_one`, which wakes only the first waiter and leaves the rest
+    /// hanging forever.
+    #[test]
+    fn request_wakes_all_concurrent_waiters() {
+        use std::{
+            future::Future,
+            pin::pin,
+            task::{Context, Poll},
+        };
+
+        let shutdown = Shutdown::new();
+        let waker = std::task::Waker::noop();
+        let mut cx = Context::from_waker(waker);
+
+        let mut w1 = pin!(shutdown.requested_signal());
+        let mut w2 = pin!(shutdown.requested_signal());
+
+        // With the flag still false, both register on the notify and suspend.
+        assert!(matches!(w1.as_mut().poll(&mut cx), Poll::Pending));
+        assert!(matches!(w2.as_mut().poll(&mut cx), Poll::Pending));
+
+        shutdown.request();
+
+        assert!(
+            matches!(w1.as_mut().poll(&mut cx), Poll::Ready(())),
+            "the first parked waiter must be woken by request()"
+        );
+        assert!(
+            matches!(w2.as_mut().poll(&mut cx), Poll::Ready(())),
+            "every parked waiter must be woken by request(), not just the first"
+        );
+    }
+}
diff --git a/bin/dipper-service/src/worker/context.rs b/bin/dipper-service/src/worker/context.rs
index 27be39a4..4b606026 100644
--- a/bin/dipper-service/src/worker/context.rs
+++ b/bin/dipper-service/src/worker/context.rs
@@ -115,6 +115,10 @@ pub struct Ctx<Q, R, C, I, T> {
     /// when the chain_listener is not configured.
     pub chain_listener_chain_id: Option<u64>,
 
+    /// Liveness watermark the worker ticks each loop iteration so the health
+    /// endpoint can detect a wedged worker.
+    pub liveness: crate::health::Liveness,
+
     /// Global reassess lock (see `ReassessLock`).
     pub reassess_lock: ReassessLock,
 
diff --git a/bin/dipper-service/src/worker/queue.rs b/bin/dipper-service/src/worker/queue.rs
index 9508ab0e..095b82b7 100644
--- a/bin/dipper-service/src/worker/queue.rs
+++ b/bin/dipper-service/src/worker/queue.rs
@@ -56,9 +56,19 @@ where
 /// A listener for the queue job available notification
 pub struct QueueImplListener(PgQueueListener);
 
-impl QueueImplListener {
-    /// Waits for a new job available notification
-    pub async fn wait_for_notification(&mut self) -> anyhow::Result<()> {
+/// A source of "a job may be available" notifications.
+///
+/// Abstracted behind a trait so the worker loop's degrade-to-polling behaviour
+/// can be unit tested without a live Postgres `LISTEN`/`NOTIFY` connection.
+#[async_trait]
+pub trait JobNotifications: Send {
+    /// Waits for the next job-available notification.
+    async fn wait_for_notification(&mut self) -> anyhow::Result<()>;
+}
+
+#[async_trait]
+impl JobNotifications for QueueImplListener {
+    async fn wait_for_notification(&mut self) -> anyhow::Result<()> {
         self.0.wait_for_notification().await
     }
 }
diff --git a/bin/dipper-service/src/worker/service.rs b/bin/dipper-service/src/worker/service.rs
index 11c8e51d..1fd2d9b7 100644
--- a/bin/dipper-service/src/worker/service.rs
+++ b/bin/dipper-service/src/worker/service.rs
@@ -12,7 +12,7 @@ use super::{
         SendIndexingAgreementProposalCtx, SubmitOfferCtx,
     },
     messages::Message,
-    queue::Queue,
+    queue::{JobNotifications, Queue},
     result::{JobError, JobResult, calculate_backoff_delay},
 };
 pub use super::{
@@ -31,6 +31,97 @@ use crate::{
 /// Default period to poll the queue for new jobs
 const DEFAULT_QUEUE_POLL_PERIOD: Duration = Duration::from_secs(1);
 
+/// Backstop timeout for processing a single job.
+///
+/// Every external call `process_job` makes is already individually bounded
+/// (IISA HTTP, indexer RPC, chain RPC + receipt polling), so the legitimate
+/// worst case is their sum — on the order of a couple of minutes. This timeout
+/// sits comfortably above that and only fires if a dependency accepts the
+/// connection but never responds, defeating the per-call timeouts. Critically,
+/// for the whole `process_job` call the job's `JobGuard` holds the row's
+/// `Running` lock (and the pgmq transaction behind it). An unbounded hang
+/// would therefore both wedge the worker loop and pin that row indefinitely.
+/// On elapse the in-flight `process_job` future is cancelled (dropped), which
+/// unblocks the loop, and the timeout is surfaced as a retryable error so the
+/// `JobGuard` reschedules the row and releases its lock. Recovery is
+/// idempotent (chain-as-source-of-truth), so re-running a job whose handler
+/// was cancelled mid-flight is safe.
+pub(crate) const PROCESS_JOB_TIMEOUT: Duration = Duration::from_secs(300);
+
+/// Base backoff for a job rescheduled after hitting [`PROCESS_JOB_TIMEOUT`].
+const JOB_TIMEOUT_RETRY_BASE_DELAY: Duration = Duration::from_secs(30);
+
+/// Runs a `process_job` future under [`PROCESS_JOB_TIMEOUT`]. On elapse it
+/// cancels the in-flight future and returns a retryable error so the worker
+/// reschedules the job (via its `JobGuard`) rather than blocking indefinitely.
+async fn run_job_with_timeout<F>(timeout: Duration, fut: F) -> JobResult<()>
+where
+    F: std::future::Future<Output = JobResult<()>>,
+{
+    match tokio::time::timeout(timeout, fut).await {
+        Ok(res) => res,
+        Err(_elapsed) => Err(JobError::Retryable(
+            anyhow::anyhow!("job processing exceeded the {timeout:?} backstop timeout"),
+            JOB_TIMEOUT_RETRY_BASE_DELAY,
+        )),
+    }
+}
+
+/// What the worker should do after waiting for the next trigger.
+#[derive(Debug, PartialEq, Eq)]
+enum Tick {
+    /// A stop signal was received; the worker loop should exit.
+    Stop,
+    /// Attempt to pop and process the next job.
+    Poll,
+}
+
+/// Waits for the next trigger to attempt a queue poll: a stop signal, a
+/// `LISTEN`/`NOTIFY` notification, or the poll-interval fallback.
+///
+/// On a listener error the listener is dropped (`*listener = None`) and the
+/// worker degrades to poll-only operation. This is correct, not a degraded
+/// state to be feared: `queue.pop()` is independent of `LISTEN`/`NOTIFY` — the
+/// notification only wakes the loop earlier than the poll interval, a latency
+/// optimisation — so a listener fault never stops job processing and never
+/// leaves the worker in an uncertain state. The caller re-subscribes on a
+/// later poll.
+async fn await_next_tick<N: JobNotifications>(
+    stop_rx: &mut watch::Receiver<bool>,
+    listener: &mut Option<N>,
+    poll_period: Duration,
+) -> Tick {
+    match listener {
+        Some(l) => {
+            tokio::select! { biased;
+                res = stop_rx.changed() => {
+                    // Sender dropped (Err) or value flipped to true: shut down.
+                    if res.is_err() || *stop_rx.borrow() { Tick::Stop } else { Tick::Poll }
+                }
+                res = l.wait_for_notification() => {
+                    if let Err(err) = res {
+                        tracing::warn!(
+                            error=?err,
+                            "job-available listener failed; degrading to poll-only until it can be re-established"
+                        );
+                        *listener = None;
+                    }
+                    Tick::Poll
+                }
+                _ = tokio::time::sleep(poll_period) => Tick::Poll,
+            }
+        }
+        None => {
+            tokio::select! { biased;
+                res = stop_rx.changed() => {
+                    if res.is_err() || *stop_rx.borrow() { Tick::Stop } else { Tick::Poll }
+                }
+                _ = tokio::time::sleep(poll_period) => Tick::Poll,
+            }
+        }
+    }
+}
+
 /// Create a worker plus a future that runs `concurrency` (>=1) loops, each
 /// draining the same queue. The queue's `FOR UPDATE SKIP LOCKED` pop hands
 /// every loop a distinct job; the default of 1 is a single serial loop.
@@ -68,6 +159,7 @@ where
         chain_listener_notify,
         bypass_chain_clock_defenses,
         chain_listener_chain_id,
+        liveness,
         reassess_lock,
         unresponsive_breaker,
         dips_accepting_cache,
@@ -111,7 +203,12 @@ where
 
         let mut set = JoinSet::new();
         for _ in 0..concurrency.max(1) {
-            set.spawn(run_loop(state.clone(), queue.clone(), stop_rx.clone()));
+            set.spawn(run_loop(
+                state.clone(),
+                queue.clone(),
+                stop_rx.clone(),
+                liveness.clone(),
+            ));
         }
         // Drop the supervisor's own receiver so `Handle::stop`'s `closed()`
         // tracks only the loops.
@@ -152,6 +249,7 @@ async fn run_loop<Q, R, C, I, T>(
     state: InnerCtx<R, WorkerQueueHandle<Q>, C, I, T>,
     queue: Q,
     mut stop_rx: watch::Receiver<bool>,
+    liveness: crate::health::Liveness,
 ) -> anyhow::Result<()>
 where
     Q: Queue<Message> + Clone + Send + Sync + 'static,
@@ -168,22 +266,36 @@ where
     I: CandidateSelection + Clone + Send + Sync + 'static,
     T: ChainClient + Clone + Send + Sync + 'static,
 {
-    let mut listener = queue.subscribe().await?;
+    // `Some` while LISTEN/NOTIFY is healthy; `None` once it has degraded to
+    // poll-only operation (see `await_next_tick`).
+    let mut listener = Some(queue.subscribe().await?);
     loop {
-        tokio::select! { biased;
-            res = stop_rx.changed() => {
-                // Sender dropped (Err) or value flipped to true: shut down.
-                if res.is_err() || *stop_rx.borrow() {
-                    return Ok(());
+        match await_next_tick(&mut stop_rx, &mut listener, DEFAULT_QUEUE_POLL_PERIOD).await {
+            Tick::Stop => return Ok(()),
+            Tick::Poll => {}
+        }
+
+        // Tick the liveness watermark on every iteration (including idle
+        // polls) so the health endpoint sees a live worker. A job that runs
+        // up to PROCESS_JOB_TIMEOUT is the longest gap between ticks.
+        liveness.record_progress();
+
+        // If the listener degraded on a previous tick, try to re-establish
+        // it. This is bounded to at most once per poll period and is
+        // non-fatal: a failure keeps us in correct poll-only mode.
+        if listener.is_none() {
+            match queue.subscribe().await {
+                Ok(l) => {
+                    tracing::info!("re-subscribed to job-available notifications");
+                    listener = Some(l);
                 }
-            }
-            res = listener.wait_for_notification() => {
-                if let Err(err) = res {
-                    tracing::error!(error=?err, "Failed to wait for job available notification");
-                    panic!("An unexpected error occurred while waiting for job available notification");
+                Err(err) => {
+                    tracing::debug!(
+                        error=?err,
+                        "job-available listener re-subscription failed; staying in poll-only mode"
+                    );
                 }
             }
-            _ = tokio::time::sleep(DEFAULT_QUEUE_POLL_PERIOD) => {}
         }
 
         // Process the job
@@ -200,7 +312,7 @@ where
         };
 
         let _span = tracing::debug_span!("process_job", job = %job.id());
-        match process_job(&state, job.desc()).await {
+        match run_job_with_timeout(PROCESS_JOB_TIMEOUT, process_job(&state, job.desc())).await {
             Ok(..) => {
                 if let Err(err) = job.remove().await {
                     tracing::debug!(error=?err, "Failed to remove job from queue");
@@ -312,3 +424,116 @@ impl<Q> Handle<Q> {
         self.stop_tx.closed().await;
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use std::future;
+
+    use async_trait::async_trait;
+    use tokio::sync::watch;
+
+    use super::{JobError, Tick, await_next_tick, run_job_with_timeout};
+    use crate::worker::queue::JobNotifications;
+
+    /// A listener stub whose `wait_for_notification` always errors. Models a
+    /// dropped/broken `LISTEN`/`NOTIFY` connection.
+    struct FailingNotifier;
+
+    #[async_trait]
+    impl JobNotifications for FailingNotifier {
+        async fn wait_for_notification(&mut self) -> anyhow::Result<()> {
+            anyhow::bail!("simulated listener failure")
+        }
+    }
+
+    /// A listener stub that never notifies (its future stays pending).
+    struct SilentNotifier;
+
+    #[async_trait]
+    impl JobNotifications for SilentNotifier {
+        async fn wait_for_notification(&mut self) -> anyhow::Result<()> {
+            future::pending().await
+        }
+    }
+
+    /// A listener error must degrade to poll-only (drop the listener) and return
+    /// `Poll`, never panic or stop. This is the regression test for the worker
+    /// `panic!` that turned a recoverable listener fault into a silent total
+    /// stall.
+    #[tokio::test]
+    async fn listener_error_degrades_to_poll_without_panicking() {
+        // Hold the sender so the stop branch stays pending.
+        let (_tx, mut rx) = watch::channel(false);
+        let mut listener = Some(FailingNotifier);
+
+        let tick =
+            await_next_tick(&mut rx, &mut listener, std::time::Duration::from_secs(60)).await;
+
+        assert_eq!(tick, Tick::Poll, "a listener error should still poll");
+        assert!(
+            listener.is_none(),
+            "a listener error should degrade to poll-only (listener dropped)"
+        );
+    }
+
+    /// A pending stop signal wins over a silent listener (biased select).
+    #[tokio::test]
+    async fn stop_signal_wins() {
+        let (tx, mut rx) = watch::channel(false);
+        tx.send(true).unwrap();
+        let mut listener = Some(SilentNotifier);
+
+        let tick =
+            await_next_tick(&mut rx, &mut listener, std::time::Duration::from_secs(60)).await;
+
+        assert_eq!(tick, Tick::Stop);
+    }
+
+    /// With no listener (already degraded), the poll-interval fallback drives
+    /// the loop so job processing continues.
+    #[tokio::test]
+    async fn poll_only_mode_ticks_on_interval() {
+        let (_tx, mut rx) = watch::channel(false);
+        let mut listener: Option<SilentNotifier> = None;
+
+        let tick =
+            await_next_tick(&mut rx, &mut listener, std::time::Duration::from_millis(10)).await;
+
+        assert_eq!(tick, Tick::Poll);
+    }
+
+    /// A job that hangs past the timeout must be turned into a retryable error
+    /// (so it reschedules and the open transaction rolls back), not awaited
+    /// indefinitely. Bounded by a test-level timeout so a regression fails fast
+    /// instead of hanging.
+    #[tokio::test]
+    async fn hung_job_times_out_as_retryable() {
+        let hung = future::pending::<super::JobResult<()>>();
+
+        let result = tokio::time::timeout(
+            std::time::Duration::from_secs(2),
+            run_job_with_timeout(std::time::Duration::from_millis(50), hung),
+        )
+        .await
+        .expect("run_job_with_timeout did not bound the hung job");
+
+        assert!(
+            matches!(result, Err(JobError::Retryable(..))),
+            "a timed-out job should be retryable, got {result:?}"
+        );
+    }
+
+    /// A job that completes within the timeout passes its result through
+    /// unchanged (including a Fatal classification).
+    #[tokio::test]
+    async fn fast_job_result_passes_through() {
+        let ok = run_job_with_timeout(std::time::Duration::from_secs(60), async { Ok(()) }).await;
+        assert!(ok.is_ok());
+
+        let fatal = run_job_with_timeout(std::time::Duration::from_secs(60), async {
+            Err(JobError::Fatal(anyhow::anyhow!("boom")))
+        })
+        .await;
+        assert!(matches!(fatal, Err(JobError::Fatal(..))));
+    }
+}
diff --git a/k8s/configmap-example.yaml b/k8s/configmap-example.yaml
index 67d465c4..7578de81 100644
--- a/k8s/configmap-example.yaml
+++ b/k8s/configmap-example.yaml
@@ -101,5 +101,9 @@ data:
         "gas_buffer_multiplier": 2.0,
         "gas_floor": 100000,
         "gas_max_addition": 200000
+      },
+      "health": {
+        "listen_addr": "0.0.0.0:8546",
+        "threshold": 600
       }
     }
diff --git a/k8s/deployment.yaml b/k8s/deployment.yaml
index a3ca0389..a7425bc6 100644
--- a/k8s/deployment.yaml
+++ b/k8s/deployment.yaml
@@ -31,15 +31,23 @@ spec:
             - name: indexer-rpc
               containerPort: 50051
               protocol: TCP
+            - name: health
+              containerPort: 8546
+              protocol: TCP
           securityContext:
             allowPrivilegeEscalation: false
             readOnlyRootFilesystem: true
             capabilities:
               drop:
                 - ALL
+          # Probe the worker health endpoint, not just a TCP socket: it returns
+          # 503 once the worker watermark goes stale, so a wedged worker (alive
+          # but making no progress) is restarted. A plain admin-rpc TCP check
+          # stays up in that state and would never catch it.
           livenessProbe:
-            tcpSocket:
-              port: admin-rpc
+            httpGet:
+              path: /health
+              port: health
             initialDelaySeconds: 30
             periodSeconds: 10
             timeoutSeconds: 5