diff --git a/crates/taskito-core/src/storage/diesel_common/logs.rs b/crates/taskito-core/src/storage/diesel_common/logs.rs
index a35d015d..9e2c0af5 100644
--- a/crates/taskito-core/src/storage/diesel_common/logs.rs
+++ b/crates/taskito-core/src/storage/diesel_common/logs.rs
@@ -43,6 +43,27 @@ macro_rules! impl_diesel_log_ops {
                 Ok(rows)
             }
 
+            /// Logs for a job with id strictly after `after_id` (cursor scan).
+            pub fn get_task_logs_after(
+                &self,
+                job_id: &str,
+                after_id: Option<&str>,
+            ) -> Result<Vec<TaskLogRow>> {
+                let mut conn = self.conn()?;
+                let mut query = task_logs::table
+                    .filter(task_logs::job_id.eq(job_id))
+                    .into_boxed();
+                if let Some(cursor) = after_id {
+                    query = query.filter(task_logs::id.gt(cursor.to_string()));
+                }
+                // UUIDv7 ids sort by creation time, so id order == time order.
+                let rows = query
+                    .order(task_logs::id.asc())
+                    .select(TaskLogRow::as_select())
+                    .load(&mut conn)?;
+                Ok(rows)
+            }
+
             /// Query logs by task name, level, etc.
             pub fn query_task_logs(
                 &self,
diff --git a/crates/taskito-core/src/storage/mod.rs b/crates/taskito-core/src/storage/mod.rs
index 6bbc9207..efd8057f 100644
--- a/crates/taskito-core/src/storage/mod.rs
+++ b/crates/taskito-core/src/storage/mod.rs
@@ -399,6 +399,13 @@ macro_rules! impl_storage {
             ) -> $crate::error::Result<Vec<$crate::storage::models::TaskLogRow>> {
                 self.get_task_logs(job_id)
             }
+            fn get_task_logs_after(
+                &self,
+                job_id: &str,
+                after_id: Option<&str>,
+            ) -> $crate::error::Result<Vec<$crate::storage::models::TaskLogRow>> {
+                self.get_task_logs_after(job_id, after_id)
+            }
             fn query_task_logs(
                 &self,
                 task_name: Option<&str>,
@@ -931,6 +938,13 @@ impl Storage for StorageBackend {
     fn get_task_logs(&self, job_id: &str) -> Result<Vec<models::TaskLogRow>> {
         delegate!(self, get_task_logs, job_id)
     }
+    fn get_task_logs_after(
+        &self,
+        job_id: &str,
+        after_id: Option<&str>,
+    ) -> Result<Vec<models::TaskLogRow>> {
+        delegate!(self, get_task_logs_after, job_id, after_id)
+    }
     fn query_task_logs(
         &self,
         task_name: Option<&str>,
diff --git a/crates/taskito-core/src/storage/redis_backend/logs.rs b/crates/taskito-core/src/storage/redis_backend/logs.rs
index 44464836..3491484d 100644
--- a/crates/taskito-core/src/storage/redis_backend/logs.rs
+++ b/crates/taskito-core/src/storage/redis_backend/logs.rs
@@ -93,6 +93,40 @@ impl RedisStorage {
         Ok(rows)
     }
 
+    /// Logs for a job with id strictly after `after_id` (cursor scan).
+    pub fn get_task_logs_after(
+        &self,
+        job_id: &str,
+        after_id: Option<&str>,
+    ) -> Result<Vec<TaskLogRow>> {
+        let mut conn = self.conn()?;
+        let by_job_key = self.key(&["logs", "by_job", job_id]);
+
+        // Filter ids before fetching values so old entries cost one cheap
+        // ZRANGEBYSCORE, not a GET per historical log.
+        let mut ids: Vec<String> = conn
+            .zrangebyscore(&by_job_key, "-inf", "+inf")
+            .map_err(map_err)?;
+        if let Some(cursor) = after_id {
+            ids.retain(|id| id.as_str() > cursor);
+        }
+
+        let mut rows = Vec::new();
+        for id in ids {
+            let log_key = self.key(&["log", &id]);
+            let data: Option<String> = conn.get(&log_key).map_err(map_err)?;
+            if let Some(d) = data {
+                let entry: LogEntry =
+                    serde_json::from_str(&d).map_err(|e| QueueError::Other(e.to_string()))?;
+                rows.push(TaskLogRow::from(entry));
+            }
+        }
+
+        // UUIDv7 ids sort by creation time, so id order == time order.
+        rows.sort_by(|a, b| a.id.cmp(&b.id));
+        Ok(rows)
+    }
+
     pub fn query_task_logs(
         &self,
         task_name: Option<&str>,
diff --git a/crates/taskito-core/src/storage/traits.rs b/crates/taskito-core/src/storage/traits.rs
index 23ee1bf1..06e75cdc 100644
--- a/crates/taskito-core/src/storage/traits.rs
+++ b/crates/taskito-core/src/storage/traits.rs
@@ -153,6 +153,9 @@ pub trait Storage: Send + Sync + Clone {
         extra: Option<&str>,
     ) -> Result<()>;
     fn get_task_logs(&self, job_id: &str) -> Result<Vec<TaskLogRow>>;
+    /// Logs for a job with id strictly after `after_id` (UUIDv7 ids are
+    /// time-ordered, so the id doubles as a stream cursor). `None` = all.
+    fn get_task_logs_after(&self, job_id: &str, after_id: Option<&str>) -> Result<Vec<TaskLogRow>>;
     fn query_task_logs(
         &self,
         task_name: Option<&str>,
diff --git a/crates/taskito-core/tests/rust/storage_tests.rs b/crates/taskito-core/tests/rust/storage_tests.rs
index 6ffefc80..5ac813f6 100644
--- a/crates/taskito-core/tests/rust/storage_tests.rs
+++ b/crates/taskito-core/tests/rust/storage_tests.rs
@@ -885,6 +885,32 @@ fn run_storage_tests(s: &impl Storage) {
     test_archived_job_payload_resolves(s);
     test_concurrent_dequeue_no_double_claim(s);
     test_rate_limit_token_exhaustion(s);
+    test_task_logs_after_cursor(s);
+}
+
+fn test_task_logs_after_cursor(s: &impl Storage) {
+    let job = s.enqueue(make_job("q-logs", "log_task")).unwrap();
+    for i in 0..3 {
+        s.write_task_log(&job.id, "log_task", "result", &format!("m{i}"), None)
+            .unwrap();
+    }
+
+    // No cursor → everything, in id (time) order, matching get_task_logs.
+    let all = s.get_task_logs_after(&job.id, None).unwrap();
+    assert_eq!(all.len(), 3);
+    assert!(all.windows(2).all(|w| w[0].id < w[1].id));
+
+    // A cursor at entry N yields only the entries written after it.
+    let after_first = s.get_task_logs_after(&job.id, Some(&all[0].id)).unwrap();
+    assert_eq!(
+        after_first
+            .iter()
+            .map(|r| r.id.as_str())
+            .collect::<Vec<_>>(),
+        all[1..].iter().map(|r| r.id.as_str()).collect::<Vec<_>>()
+    );
+    let after_last = s.get_task_logs_after(&job.id, Some(&all[2].id)).unwrap();
+    assert!(after_last.is_empty());
 }
 
 fn test_rate_limit_token_exhaustion(s: &impl Storage) {
diff --git a/crates/taskito-node/src/dispatcher.rs b/crates/taskito-node/src/dispatcher.rs
index 873a126d..6223ff42 100644
--- a/crates/taskito-node/src/dispatcher.rs
+++ b/crates/taskito-node/src/dispatcher.rs
@@ -8,7 +8,7 @@
 use std::time::{Duration, Instant};
 
 use crossbeam_channel::Sender;
-use napi::bindgen_prelude::{spawn, Buffer, Promise};
+use napi::bindgen_prelude::{spawn, spawn_blocking, Buffer, Promise};
 use napi::threadsafe_function::{ErrorStrategy, ThreadsafeFunction, ThreadsafeFunctionCallMode};
 use taskito_core::job::Job;
 use taskito_core::scheduler::JobResult;
@@ -48,8 +48,19 @@ impl WorkerDispatcher for NodeDispatcher {
             let storage = self.storage.clone();
             let result_tx = result_tx.clone();
             spawn(async move {
+                let job_id = job.id.clone();
                 let result = run_one(&callback, &storage, job).await;
-                let _ = result_tx.send(result);
+                // A full bounded channel parks the sender — do it on the
+                // blocking pool, never on the shared async runtime.
+                match spawn_blocking(move || result_tx.send(result)).await {
+                    Ok(Ok(())) => {}
+                    Ok(Err(_)) | Err(_) => {
+                        // A closed channel means the drain loop died — surface it.
+                        log::error!(
+                            "[taskito-node] result channel closed; dropping outcome for {job_id}"
+                        );
+                    }
+                }
             });
         }
     }
@@ -58,12 +69,14 @@ impl WorkerDispatcher for NodeDispatcher {
 }
 
 /// Invoke the JS task for one job and translate the outcome into a [`JobResult`].
-async fn run_one(callback: &TaskCallback, storage: &StorageBackend, job: Job) -> JobResult {
+async fn run_one(callback: &TaskCallback, storage: &StorageBackend, mut job: Job) -> JobResult {
     let started = Instant::now();
     let invocation = JsTaskInvocation {
         id: job.id.clone(),
         task_name: job.task_name.clone(),
-        payload: Buffer::from(job.payload.clone()),
+        // Never read from `job` again — take, don't clone (payloads can be
+        // large); `job` stays whole for the later `failure(job, ...)` moves.
+        payload: Buffer::from(std::mem::take(&mut job.payload)),
     };
 
     // The callback runs on the JS thread and returns a Promise; bridge its
diff --git a/crates/taskito-node/src/error.rs b/crates/taskito-node/src/error.rs
index 42882f53..fba6cf60 100644
--- a/crates/taskito-node/src/error.rs
+++ b/crates/taskito-node/src/error.rs
@@ -8,6 +8,14 @@ pub fn to_napi_err(err: QueueError) -> Error {
     Error::new(Status::GenericFailure, err.to_string())
 }
 
+/// Map a blocking-task join failure onto a napi [`Error`].
+pub fn join_to_napi_err(err: tokio::task::JoinError) -> Error {
+    Error::new(
+        Status::GenericFailure,
+        format!("blocking storage task failed: {err}"),
+    )
+}
+
 /// Build an `InvalidArg` error for caller-supplied input that fails validation
 /// at the N-API boundary (negative pagination, zero pool size, bad rate limit…).
 pub fn invalid_arg(message: impl Into<String>) -> Error {
diff --git a/crates/taskito-node/src/queue/admin.rs b/crates/taskito-node/src/queue/admin.rs
index 48ee0712..93fa54ba 100644
--- a/crates/taskito-node/src/queue/admin.rs
+++ b/crates/taskito-node/src/queue/admin.rs
@@ -1,14 +1,16 @@
-//! Management (mutating) methods on `JsQueue`.
+//! Management (mutating) methods on `JsQueue`. List/purge methods scan whole
+//! tables, so they are async and offload to the blocking pool; single-row
+//! operations stay sync.
 
 use std::collections::HashMap;
 
-use napi::bindgen_prelude::Result;
+use napi::bindgen_prelude::{spawn_blocking, Result};
 use napi_derive::napi;
 use taskito_core::Storage;
 
 use super::JsQueue;
 use crate::convert::{dead_job_to_js, JsDeadJob};
-use crate::error::{non_negative, to_napi_err};
+use crate::error::{join_to_napi_err, non_negative, to_napi_err};
 
 const DEFAULT_LIMIT: i64 = 50;
 
@@ -16,16 +18,25 @@ const DEFAULT_LIMIT: i64 = 50;
 impl JsQueue {
     /// List dead-letter entries (paginated).
     #[napi]
-    pub fn dead_letters(&self, limit: Option<i64>, offset: Option<i64>) -> Result<Vec<JsDeadJob>> {
+    pub async fn dead_letters(
+        &self,
+        limit: Option<i64>,
+        offset: Option<i64>,
+    ) -> Result<Vec<JsDeadJob>> {
         let limit = non_negative(limit.unwrap_or(DEFAULT_LIMIT), "limit")?;
         let offset = non_negative(offset.unwrap_or(0), "offset")?;
-        let dead = self.storage.list_dead(limit, offset).map_err(to_napi_err)?;
-        Ok(dead.into_iter().map(dead_job_to_js).collect())
+        let storage = self.storage.clone();
+        spawn_blocking(move || {
+            let dead = storage.list_dead(limit, offset).map_err(to_napi_err)?;
+            Ok(dead.into_iter().map(dead_job_to_js).collect())
+        })
+        .await
+        .map_err(join_to_napi_err)?
     }
 
     /// List dead-letter entries for a single task (paginated, newest first).
     #[napi]
-    pub fn dead_letters_by_task(
+    pub async fn dead_letters_by_task(
         &self,
         task_name: String,
         limit: Option<i64>,
@@ -33,20 +44,29 @@ impl JsQueue {
     ) -> Result<Vec<JsDeadJob>> {
         let limit = non_negative(limit.unwrap_or(DEFAULT_LIMIT), "limit")?;
         let offset = non_negative(offset.unwrap_or(0), "offset")?;
-        let dead = self
-            .storage
-            .list_dead_by_task(&task_name, limit, offset)
-            .map_err(to_napi_err)?;
-        Ok(dead.into_iter().map(dead_job_to_js).collect())
+        let storage = self.storage.clone();
+        spawn_blocking(move || {
+            let dead = storage
+                .list_dead_by_task(&task_name, limit, offset)
+                .map_err(to_napi_err)?;
+            Ok(dead.into_iter().map(dead_job_to_js).collect())
+        })
+        .await
+        .map_err(join_to_napi_err)?
     }
 
     /// Delete every dead-letter entry for a task. Returns the count removed.
     #[napi]
-    pub fn purge_dead_by_task(&self, task_name: String) -> Result<i64> {
-        self.storage
-            .purge_dead_by_task(&task_name)
-            .map(|n| n as i64)
-            .map_err(to_napi_err)
+    pub async fn purge_dead_by_task(&self, task_name: String) -> Result<i64> {
+        let storage = self.storage.clone();
+        spawn_blocking(move || {
+            storage
+                .purge_dead_by_task(&task_name)
+                .map(|n| n as i64)
+                .map_err(to_napi_err)
+        })
+        .await
+        .map_err(join_to_napi_err)?
     }
 
     /// Re-enqueue a dead-letter entry. Returns the new job id.
@@ -63,22 +83,32 @@ impl JsQueue {
 
     /// Purge dead-letter entries older than `older_than_ms`. Returns the count removed.
     #[napi]
-    pub fn purge_dead(&self, older_than_ms: i64) -> Result<i64> {
+    pub async fn purge_dead(&self, older_than_ms: i64) -> Result<i64> {
         let older_than_ms = non_negative(older_than_ms, "olderThanMs")?;
-        self.storage
-            .purge_dead(older_than_ms)
-            .map(|n| n as i64)
-            .map_err(to_napi_err)
+        let storage = self.storage.clone();
+        spawn_blocking(move || {
+            storage
+                .purge_dead(older_than_ms)
+                .map(|n| n as i64)
+                .map_err(to_napi_err)
+        })
+        .await
+        .map_err(join_to_napi_err)?
     }
 
     /// Purge completed jobs older than `older_than_ms`. Returns the count removed.
     #[napi]
-    pub fn purge_completed(&self, older_than_ms: i64) -> Result<i64> {
+    pub async fn purge_completed(&self, older_than_ms: i64) -> Result<i64> {
         let older_than_ms = non_negative(older_than_ms, "olderThanMs")?;
-        self.storage
-            .purge_completed(older_than_ms)
-            .map(|n| n as i64)
-            .map_err(to_napi_err)
+        let storage = self.storage.clone();
+        spawn_blocking(move || {
+            storage
+                .purge_completed(older_than_ms)
+                .map(|n| n as i64)
+                .map_err(to_napi_err)
+        })
+        .await
+        .map_err(join_to_napi_err)?
     }
 
     /// Pause a queue — workers stop dispatching its jobs until resumed.
diff --git a/crates/taskito-node/src/queue/inspect.rs b/crates/taskito-node/src/queue/inspect.rs
index 6dc60d7b..ed76ecd8 100644
--- a/crates/taskito-node/src/queue/inspect.rs
+++ b/crates/taskito-node/src/queue/inspect.rs
@@ -1,8 +1,10 @@
-//! Read-only inspection methods on `JsQueue`.
+//! Read-only inspection methods on `JsQueue`. These scan/aggregate over
+//! storage, so each is async and offloads the blocking I/O to the blocking
+//! pool instead of stalling the JS event loop for the DB round-trip.
 
 use std::collections::HashMap;
 
-use napi::bindgen_prelude::Result;
+use napi::bindgen_prelude::{spawn_blocking, Result};
 use napi_derive::napi;
 use taskito_core::Storage;
 
@@ -12,7 +14,7 @@ use crate::convert::{
     job_error_to_js, job_to_js, metric_to_js, stats_to_js, status_code, worker_to_js, JsJob,
     JsJobError, JsMetric, JsStats, JsWorkerRow,
 };
-use crate::error::{invalid_arg, non_negative, to_napi_err};
+use crate::error::{invalid_arg, join_to_napi_err, non_negative, to_napi_err};
 
 const DEFAULT_LIMIT: i64 = 50;
 
@@ -20,29 +22,42 @@ const DEFAULT_LIMIT: i64 = 50;
 impl JsQueue {
     /// Job counts by status across all queues.
     #[napi]
-    pub fn stats(&self) -> Result<JsStats> {
-        self.storage.stats().map(stats_to_js).map_err(to_napi_err)
+    pub async fn stats(&self) -> Result<JsStats> {
+        let storage = self.storage.clone();
+        spawn_blocking(move || storage.stats().map(stats_to_js).map_err(to_napi_err))
+            .await
+            .map_err(join_to_napi_err)?
     }
 
     /// Job counts by status for a single queue.
     #[napi]
-    pub fn stats_by_queue(&self, queue: String) -> Result<JsStats> {
-        self.storage
-            .stats_by_queue(&queue)
-            .map(stats_to_js)
-            .map_err(to_napi_err)
+    pub async fn stats_by_queue(&self, queue: String) -> Result<JsStats> {
+        let storage = self.storage.clone();
+        spawn_blocking(move || {
+            storage
+                .stats_by_queue(&queue)
+                .map(stats_to_js)
+                .map_err(to_napi_err)
+        })
+        .await
+        .map_err(join_to_napi_err)?
     }
 
     /// Job counts by status, keyed by queue name.
     #[napi]
-    pub fn stats_all_queues(&self) -> Result<HashMap<String, JsStats>> {
-        let all = self.storage.stats_all_queues().map_err(to_napi_err)?;
-        Ok(all.into_iter().map(|(k, v)| (k, stats_to_js(v))).collect())
+    pub async fn stats_all_queues(&self) -> Result<HashMap<String, JsStats>> {
+        let storage = self.storage.clone();
+        spawn_blocking(move || {
+            let all = storage.stats_all_queues().map_err(to_napi_err)?;
+            Ok(all.into_iter().map(|(k, v)| (k, stats_to_js(v))).collect())
+        })
+        .await
+        .map_err(join_to_napi_err)?
     }
 
     /// List jobs, optionally filtered by status/queue/task and paginated.
     #[napi]
-    pub fn list_jobs(&self, filter: Option<JobFilter>) -> Result<Vec<JsJob>> {
+    pub async fn list_jobs(&self, filter: Option<JobFilter>) -> Result<Vec<JsJob>> {
         let filter = filter.unwrap_or_default();
         // An unrecognized status would otherwise silently widen the result set
         // to every job; reject it so a typo fails loudly.
@@ -55,41 +70,60 @@ impl JsQueue {
         };
         let limit = non_negative(filter.limit.unwrap_or(DEFAULT_LIMIT), "limit")?;
         let offset = non_negative(filter.offset.unwrap_or(0), "offset")?;
-        let jobs = self
-            .storage
-            .list_jobs(
-                status,
-                filter.queue.as_deref(),
-                filter.task.as_deref(),
-                limit,
-                offset,
-                self.namespace.as_deref(),
-            )
-            .map_err(to_napi_err)?;
-        Ok(jobs.into_iter().map(job_to_js).collect())
+        let storage = self.storage.clone();
+        let namespace = self.namespace.clone();
+        spawn_blocking(move || {
+            let jobs = storage
+                .list_jobs(
+                    status,
+                    filter.queue.as_deref(),
+                    filter.task.as_deref(),
+                    limit,
+                    offset,
+                    namespace.as_deref(),
+                )
+                .map_err(to_napi_err)?;
+            Ok(jobs.into_iter().map(job_to_js).collect())
+        })
+        .await
+        .map_err(join_to_napi_err)?
     }
 
     /// Error history for a job (one entry per failed attempt).
     #[napi]
-    pub fn get_job_errors(&self, job_id: String) -> Result<Vec<JsJobError>> {
-        let errors = self.storage.get_job_errors(&job_id).map_err(to_napi_err)?;
-        Ok(errors.into_iter().map(job_error_to_js).collect())
+    pub async fn get_job_errors(&self, job_id: String) -> Result<Vec<JsJobError>> {
+        let storage = self.storage.clone();
+        spawn_blocking(move || {
+            let errors = storage.get_job_errors(&job_id).map_err(to_napi_err)?;
+            Ok(errors.into_iter().map(job_error_to_js).collect())
+        })
+        .await
+        .map_err(join_to_napi_err)?
     }
 
     /// Per-execution task metrics within the last `since_ms` milliseconds.
     #[napi]
-    pub fn get_metrics(&self, task: Option<String>, since_ms: i64) -> Result<Vec<JsMetric>> {
-        let metrics = self
-            .storage
-            .get_metrics(task.as_deref(), since_ms)
-            .map_err(to_napi_err)?;
-        Ok(metrics.into_iter().map(metric_to_js).collect())
+    pub async fn get_metrics(&self, task: Option<String>, since_ms: i64) -> Result<Vec<JsMetric>> {
+        let storage = self.storage.clone();
+        spawn_blocking(move || {
+            let metrics = storage
+                .get_metrics(task.as_deref(), since_ms)
+                .map_err(to_napi_err)?;
+            Ok(metrics.into_iter().map(metric_to_js).collect())
+        })
+        .await
+        .map_err(join_to_napi_err)?
     }
 
     /// List registered workers (heartbeat + identity).
     #[napi]
-    pub fn list_workers(&self) -> Result<Vec<JsWorkerRow>> {
-        let workers = self.storage.list_workers().map_err(to_napi_err)?;
-        Ok(workers.into_iter().map(worker_to_js).collect())
+    pub async fn list_workers(&self) -> Result<Vec<JsWorkerRow>> {
+        let storage = self.storage.clone();
+        spawn_blocking(move || {
+            let workers = storage.list_workers().map_err(to_napi_err)?;
+            Ok(workers.into_iter().map(worker_to_js).collect())
+        })
+        .await
+        .map_err(join_to_napi_err)?
     }
 }
diff --git a/crates/taskito-node/src/queue/logs.rs b/crates/taskito-node/src/queue/logs.rs
index 834f012a..5e8be8cb 100644
--- a/crates/taskito-node/src/queue/logs.rs
+++ b/crates/taskito-node/src/queue/logs.rs
@@ -33,4 +33,19 @@ impl JsQueue {
         let rows = self.storage.get_task_logs(&job_id).map_err(to_napi_err)?;
         Ok(rows.into_iter().map(log_to_js).collect())
     }
+
+    /// Task-log entries for a job with id after `afterId` (UUIDv7 ids are
+    /// time-ordered, so the last seen id is the stream cursor).
+    #[napi]
+    pub fn get_task_logs_after(
+        &self,
+        job_id: String,
+        after_id: Option<String>,
+    ) -> Result<Vec<JsTaskLog>> {
+        let rows = self
+            .storage
+            .get_task_logs_after(&job_id, after_id.as_deref())
+            .map_err(to_napi_err)?;
+        Ok(rows.into_iter().map(log_to_js).collect())
+    }
 }
diff --git a/crates/taskito-node/src/worker.rs b/crates/taskito-node/src/worker.rs
index c9c68175..0c24fa65 100644
--- a/crates/taskito-node/src/worker.rs
+++ b/crates/taskito-node/src/worker.rs
@@ -179,16 +179,22 @@ pub fn start_worker(
     let scheduler_results = scheduler;
     spawn_blocking(move || {
         while let Ok(result) = result_rx.recv() {
-            match scheduler_results.handle_result(result) {
+            // A panicking result must not kill the drain loop — a dead loop
+            // silently drops every later outcome.
+            let handled = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
+                scheduler_results.handle_result(result)
+            }));
+            match handled {
                 // Surface each outcome to JS so the shell can emit events and run
                 // middleware (the events layer).
-                Ok(outcome) => {
+                Ok(Ok(outcome)) => {
                     outcome_callback.call(
                         outcome_to_js(&outcome),
                         ThreadsafeFunctionCallMode::NonBlocking,
                     );
                 }
-                Err(err) => log::error!("[taskito-node] result handling error: {err}"),
+                Ok(Err(err)) => log::error!("[taskito-node] result handling error: {err}"),
+                Err(_) => log::error!("[taskito-node] result handling panicked; outcome dropped"),
             }
         }
     });
@@ -211,35 +217,51 @@ fn spawn_worker_lifecycle(
 ) {
     let hostname = gethostname::gethostname().to_string_lossy().to_string();
     let pid = std::process::id() as i32;
-    // Log lifecycle failures: a worker that can't register/heartbeat goes
-    // invisible or stale in the dashboard, and a silent error hides that.
-    if let Err(err) = storage.register_worker(
-        &worker_id,
-        &queues_csv,
-        None,
-        None,
-        None,
-        capacity as i32,
-        Some(&hostname),
-        Some(pid),
-        Some("node"),
-    ) {
-        log::warn!("[taskito-node] worker registration failed: {err}");
-    }
-
+    // Lifecycle calls are blocking storage I/O → run each on the blocking
+    // pool, off the JS thread and the shared async runtime. Failures are
+    // logged: a silent one leaves the worker invisible in the dashboard.
     spawn(async move {
+        let reg_storage = storage.clone();
+        let reg_id = worker_id.clone();
+        match spawn_blocking(move || {
+            reg_storage.register_worker(
+                &reg_id,
+                &queues_csv,
+                None,
+                None,
+                None,
+                capacity as i32,
+                Some(&hostname),
+                Some(pid),
+                Some("node"),
+            )
+        })
+        .await
+        {
+            Ok(Err(err)) => log::warn!("[taskito-node] worker registration failed: {err}"),
+            Err(err) => log::warn!("[taskito-node] worker registration task failed: {err}"),
+            Ok(Ok(_)) => {}
+        }
+
         loop {
             tokio::select! {
                 _ = stop.notified() => break,
                 _ = tokio::time::sleep(HEARTBEAT_INTERVAL) => {
-                    if let Err(err) = storage.heartbeat(&worker_id, None) {
-                        log::warn!("[taskito-node] worker heartbeat failed: {err}");
+                    let hb_storage = storage.clone();
+                    let hb_id = worker_id.clone();
+                    match spawn_blocking(move || hb_storage.heartbeat(&hb_id, None)).await {
+                        Ok(Err(err)) => log::warn!("[taskito-node] worker heartbeat failed: {err}"),
+                        Err(err) => log::warn!("[taskito-node] worker heartbeat task failed: {err}"),
+                        Ok(Ok(_)) => {}
                     }
                 }
             }
         }
-        if let Err(err) = storage.unregister_worker(&worker_id) {
-            log::warn!("[taskito-node] worker unregister failed: {err}");
+
+        match spawn_blocking(move || storage.unregister_worker(&worker_id)).await {
+            Ok(Err(err)) => log::warn!("[taskito-node] worker unregister failed: {err}"),
+            Err(err) => log::warn!("[taskito-node] worker unregister task failed: {err}"),
+            Ok(Ok(_)) => {}
         }
     });
 }
diff --git a/docs/content/docs/node/api-reference/queue.mdx b/docs/content/docs/node/api-reference/queue.mdx
index df34c064..80f29143 100644
--- a/docs/content/docs/node/api-reference/queue.mdx
+++ b/docs/content/docs/node/api-reference/queue.mdx
@@ -43,16 +43,19 @@ const queue = new Queue(options);
 
 ## Inspection & management
 
+Scan-heavy calls (marked `Promise` below) run off the JS event loop on a
+background thread pool.
+
 | Method | Description |
 |---|---|
-| `stats()` / `statsByQueue(q)` / `statsAllQueues()` | Counts by status. |
-| `listJobs(filter)` | List jobs. |
-| `getJobErrors(id)` | Per-attempt error history. |
-| `getMetrics(windowMs, task?)` | Aggregated metrics. |
-| `deadLetters()` / `retryDead(id)` / `deleteDead(id)` / `purgeDead(ms)` | DLQ ops. |
-| `purgeCompleted(ms)` | Drop old completed jobs. |
+| `stats()` → `Promise<Stats>` / `statsByQueue(q)` → `Promise<Stats>` / `statsAllQueues()` → `Promise<Record<string, Stats>>` | Counts by status. |
+| `listJobs(filter)` → `Promise<Job[]>` | List jobs. |
+| `getJobErrors(id)` → `Promise<JobError[]>` | Per-attempt error history. |
+| `getMetrics(windowMs, task?)` → `Promise<Metric[]>` | Aggregated metrics. |
+| `deadLetters()` → `Promise<DeadJob[]>` / `retryDead(id)` / `deleteDead(id)` / `purgeDead(ms)` → `Promise<number>` | DLQ ops. |
+| `purgeCompleted(ms)` → `Promise<number>` | Drop old completed jobs. |
 | `pauseQueue(q)` / `resumeQueue(q)` / `listPausedQueues()` | Queue control. |
-| `listWorkers()` | Registered workers. |
+| `listWorkers()` → `Promise<WorkerInfo[]>` | Registered workers. |
 
 ## Subsystems
 
diff --git a/docs/content/docs/node/guides/extensibility/webhooks.mdx b/docs/content/docs/node/guides/extensibility/webhooks.mdx
index 04bd0158..91dfa80e 100644
--- a/docs/content/docs/node/guides/extensibility/webhooks.mdx
+++ b/docs/content/docs/node/guides/extensibility/webhooks.mdx
@@ -38,8 +38,34 @@ queue.webhooks.delete(hook.id);
 Deliveries fire from the **worker process** (where events originate). Verify the
 signature on your endpoint by HMAC-ing the raw body with the shared secret.
 
+## Delivery log
+
+`queue.webhooks.deliveries(id)` returns the most recent attempt-chains for a
+webhook (bounded, newest last):
+
+```ts
+for (const delivery of queue.webhooks.deliveries(hook.id)) {
+  console.log(delivery.status, delivery.responseCode, delivery.attempts);
+}
+```
+
+| Field | Description |
+|---|---|
+| `id` / `webhookId` / `event` | Identifiers for the delivery and its subscription. |
+| `status` | `"delivered"` (2xx) or `"dead"` (retries exhausted). |
+| `ok` | `true` when `status` is `"delivered"`. |
+| `attempts` | Number of HTTP attempts made. |
+| `payload` | The JSON body that was POSTed. |
+| `taskName` / `jobId` | The job that triggered the event, if any. |
+| `responseCode` | Last HTTP status, or `null` on a network error. |
+| `responseBody` | Last response body, truncated to 2 KiB, or `null`. |
+| `latencyMs` | Time from first attempt to the final outcome. |
+| `error` | The last error message, if `status` is `"dead"`. |
+| `createdAt` / `completedAt` | Unix milliseconds. |
+
 <Callout type="info">
   The [dashboard](/node/guides/operations/dashboard) exposes `/api/webhooks` for
-  managing hooks. Failed deliveries retry with backoff and survive restarts
-  since they are persisted in storage.
+  managing hooks, and the same delivery log — paged and in snake_case — at
+  [`/api/webhooks/:id/deliveries`](/node/guides/operations/dashboard-api).
+  Webhook subscriptions are persisted in storage and survive restarts.
 </Callout>
diff --git a/docs/content/docs/node/guides/integrations/express.mdx b/docs/content/docs/node/guides/integrations/express.mdx
index 94644969..fec646a0 100644
--- a/docs/content/docs/node/guides/integrations/express.mdx
+++ b/docs/content/docs/node/guides/integrations/express.mdx
@@ -31,7 +31,7 @@ JSON body parsing is mounted internally on the router.
 | `GET` | `/stats/queues?queue=` | Per-queue statistics. |
 | `GET` | `/jobs/:id` | Fetch a job by ID. |
 | `GET` | `/jobs/:id/errors` | Error history for a job. |
-| `GET` | `/jobs/:id/result?timeoutMs=` | Wait for a result (long-poll). |
+| `GET` | `/jobs/:id/result?timeoutMs=` | Wait for a result (long-poll). `timeoutMs` is clamped to `resultTimeoutMs`. |
 | `POST` | `/jobs/:id/cancel` | Request cancellation. |
 | `GET` | `/dead-letters?limit=&offset=` | List dead-lettered jobs. |
 | `POST` | `/dead-letters/:id/retry` | Retry a dead-lettered job. |
@@ -42,7 +42,7 @@ JSON body parsing is mounted internally on the router.
 |---|---|---|---|
 | `includeRoutes` | `string[]` | all | Route names to expose (see below). |
 | `excludeRoutes` | `string[]` | none | Route names to suppress. |
-| `resultTimeoutMs` | `number` | `5000` | Max wait for `/jobs/:id/result`. |
+| `resultTimeoutMs` | `number` | `5000` | Ceiling for `/jobs/:id/result`; a client `timeoutMs` above this is clamped down to it. |
 
 Route names: `enqueue`, `stats`, `queue-stats`, `job`, `job-errors`,
 `job-result`, `cancel`, `dead-letters`, `retry-dead`.
diff --git a/docs/content/docs/node/guides/integrations/nest.mdx b/docs/content/docs/node/guides/integrations/nest.mdx
index 61acea83..e4ff5622 100644
--- a/docs/content/docs/node/guides/integrations/nest.mdx
+++ b/docs/content/docs/node/guides/integrations/nest.mdx
@@ -35,6 +35,10 @@ export class ReportService {
       timeoutMs: 60_000,
     });
   }
+
+  async health() {
+    return this.tasks.stats(); // Promise<Stats>
+  }
 }
 ```
 
@@ -45,9 +49,9 @@ export class ReportService {
 | `enqueue(task, args?, options?)` | Enqueue a job, returns the job ID. |
 | `result(id, options?)` | Wait for a result. |
 | `getJob(id)` | Fetch job details by ID. |
-| `stats()` | Overall queue statistics. |
+| `stats()` | Overall queue statistics. Returns a `Promise`. |
 | `requestCancel(id)` | Request cancellation of a running job. |
-| `deadLetters(limit?, offset?)` | List dead-lettered jobs. |
+| `deadLetters(limit?, offset?)` | List dead-lettered jobs. Returns a `Promise`. |
 | `queue` | The raw `Queue` instance for the full API surface. |
 
 <Callout type="info">
diff --git a/docs/content/docs/node/guides/observability/monitoring.mdx b/docs/content/docs/node/guides/observability/monitoring.mdx
index 560c30d1..95b0459a 100644
--- a/docs/content/docs/node/guides/observability/monitoring.mdx
+++ b/docs/content/docs/node/guides/observability/monitoring.mdx
@@ -5,16 +5,17 @@ description: "Queue stats, per-task metrics, job progress, and worker heartbeats
 
 Read live queue health straight from the `Queue` handle — no extra service. The
 same data powers the [dashboard](/node/guides/operations/dashboard) and the
-[CLI](/node/guides/operations/cli) read commands.
+[CLI](/node/guides/operations/cli) read commands. These scan-heavy methods run
+off the JS event loop on a background thread pool and return a `Promise`.
 
 ## Stats
 
 Counts by status, globally or per queue:
 
 ```ts
-queue.stats(); // { pending, running, completed, failed, dead, cancelled }
-queue.statsByQueue("default");
-queue.statsAllQueues(); // Record<queue, stats>
+await queue.stats(); // { pending, running, completed, failed, dead, cancelled }
+await queue.statsByQueue("default");
+await queue.statsAllQueues(); // Record<queue, stats>
 ```
 
 ## Metrics
@@ -24,8 +25,8 @@ into per-task counts, a success/failure split, and latency percentiles
 (p50/p95/p99):
 
 ```ts
-queue.getMetrics(3_600_000); // last hour, all tasks
-queue.getMetrics(3_600_000, "add"); // one task
+await queue.getMetrics(3_600_000); // last hour, all tasks
+await queue.getMetrics(3_600_000, "add"); // one task
 ```
 
 ## Progress
@@ -51,7 +52,7 @@ Each running worker registers and heartbeats every 5 seconds; list the live
 fleet:
 
 ```ts
-queue.listWorkers();
+await queue.listWorkers();
 // [{ workerId, hostname, pid, queues, status, lastHeartbeat, threads, ... }]
 ```
 
diff --git a/docs/content/docs/node/guides/operations/dashboard-api.mdx b/docs/content/docs/node/guides/operations/dashboard-api.mdx
index 2d49d327..d3afdc81 100644
--- a/docs/content/docs/node/guides/operations/dashboard-api.mdx
+++ b/docs/content/docs/node/guides/operations/dashboard-api.mdx
@@ -23,7 +23,41 @@ directly too. All timestamps are Unix milliseconds.
 | `GET /api/event-types` | Known job event names. |
 | `GET /api/workflows/runs` · `/:id` | Workflow runs, and one run. |
 | `GET /api/workflows/runs/:id/dag` · `/children` | A run's DAG graph and child runs. |
-| `GET /api/webhooks` · `/:id` · `/:id/deliveries` | Webhook config and delivery log. |
+| `GET /api/webhooks` · `/:id` | Webhook config. |
+| `GET /api/webhooks/:id/deliveries` | Paged delivery log — see below. |
+
+### Webhook deliveries
+
+`GET /api/webhooks/:id/deliveries` supports `?status=&limit=&offset=` and
+returns a page, fields in snake_case:
+
+```json
+{
+  "items": [
+    {
+      "id": "d_1",
+      "subscription_id": "wh_1",
+      "event": "job.dead",
+      "payload": { "jobId": "j_1", "taskName": "sendEmail" },
+      "task_name": "sendEmail",
+      "job_id": "j_1",
+      "status": "delivered",
+      "attempts": 1,
+      "response_code": 200,
+      "response_body": "ok",
+      "latency_ms": 42,
+      "error": null,
+      "created_at": 1732550400000,
+      "completed_at": 1732550400042
+    }
+  ],
+  "total": 128,
+  "limit": 50,
+  "offset": 0
+}
+```
+
+`status` is `"delivered"` or `"dead"`. `response_body` is truncated to 2 KiB.
 
 ## Write
 
diff --git a/docs/content/docs/node/guides/operations/inspection.mdx b/docs/content/docs/node/guides/operations/inspection.mdx
index f2b2b198..3b23c189 100644
--- a/docs/content/docs/node/guides/operations/inspection.mdx
+++ b/docs/content/docs/node/guides/operations/inspection.mdx
@@ -10,20 +10,26 @@ queue-wide health — stats, per-task metrics, progress, workers — see
 ## Jobs
 
 ```ts
-queue.listJobs({ status: "failed", limit: 50 });
-queue.getJobErrors(id); // per-attempt error history
+await queue.listJobs({ status: "failed", limit: 50 });
+await queue.getJobErrors(id); // per-attempt error history
 ```
 
 ## Dead letters
 
 ```ts
-queue.deadLetters();
+await queue.deadLetters();
 queue.retryDead(deadId);
 queue.deleteDead(deadId);
-queue.purgeDead(olderThanMs);
-queue.purgeCompleted(olderThanMs);
+await queue.purgeDead(olderThanMs);
+await queue.purgeCompleted(olderThanMs);
 ```
 
+<Callout type="info">
+  Scan-heavy methods (`listJobs`, `getJobErrors`, `deadLetters`, `purgeDead`,
+  `purgeCompleted`) run off the JS event loop on a background thread pool and
+  return a `Promise`.
+</Callout>
+
 ## Pause / resume
 
 ```ts
diff --git a/docs/content/docs/node/guides/operations/migration.mdx b/docs/content/docs/node/guides/operations/migration.mdx
index 949c7953..bc415f66 100644
--- a/docs/content/docs/node/guides/operations/migration.mdx
+++ b/docs/content/docs/node/guides/operations/migration.mdx
@@ -22,7 +22,7 @@ handlers are **registered by name** on the queue rather than wired to a separate
 | `QueueEvents` | `queue.on(event, …)` + [middleware](/node/guides/extensibility/middleware) |
 | Repeatable jobs (cron) | `queue.registerPeriodic(name, task, cron)` |
 | Flows (parent/child) | [Workflows](/node/guides/workflows) (DAG, fan-out, gates, saga) |
-| `queue.getJobCounts()` | `queue.stats()` / `statsAllQueues()` |
+| `queue.getJobCounts()` | `await queue.stats()` / `statsAllQueues()` |
 | Bull Board | the built-in [dashboard](/node/guides/operations/dashboard) |
 
 ## Side by side
diff --git a/docs/content/docs/node/guides/operations/troubleshooting.mdx b/docs/content/docs/node/guides/operations/troubleshooting.mdx
index bef9af10..d844a0df 100644
--- a/docs/content/docs/node/guides/operations/troubleshooting.mdx
+++ b/docs/content/docs/node/guides/operations/troubleshooting.mdx
@@ -59,6 +59,6 @@ in-flight job. Offload CPU-bound work to `worker_threads` and `await` it, or set
 a [`timeoutMs`](/node/guides/reliability/timeouts) so a stuck attempt is aborted.
 
 <Callout type="info">
-  Inspect failures with `queue.getJobErrors(id)` and `queue.deadLetters()`; see
+  Inspect failures with `await queue.getJobErrors(id)` and `await queue.deadLetters()`; see
   [job management](/node/guides/operations/inspection).
 </Callout>
diff --git a/docs/content/docs/node/guides/reliability/dead-letter.mdx b/docs/content/docs/node/guides/reliability/dead-letter.mdx
index c502003f..20da7899 100644
--- a/docs/content/docs/node/guides/reliability/dead-letter.mdx
+++ b/docs/content/docs/node/guides/reliability/dead-letter.mdx
@@ -8,16 +8,16 @@ the `job.dead` [event](/node/guides/extensibility/events) fires. The DLQ keeps
 the payload and error history so you can investigate and replay.
 
 ```ts
-const dead = queue.deadLetters();   // dead-letter entries
-queue.retryDead(deadId);            // re-enqueue (preserves notes/metadata)
-queue.deleteDead(deadId);           // drop one entry
-queue.purgeDead(olderThanMs);       // bulk-drop entries older than a cutoff
+const dead = await queue.deadLetters(); // dead-letter entries
+queue.retryDead(deadId);                // re-enqueue (preserves notes/metadata)
+queue.deleteDead(deadId);                // drop one entry
+await queue.purgeDead(olderThanMs);      // bulk-drop entries older than a cutoff
 ```
 
 Inspect why a job failed:
 
 ```ts
-queue.getJobErrors(jobId); // per-attempt error history
+await queue.getJobErrors(jobId); // per-attempt error history
 ```
 
 From the CLI:
diff --git a/docs/content/docs/node/guides/reliability/error-handling.mdx b/docs/content/docs/node/guides/reliability/error-handling.mdx
index f0e35b7b..38ce8d3c 100644
--- a/docs/content/docs/node/guides/reliability/error-handling.mdx
+++ b/docs/content/docs/node/guides/reliability/error-handling.mdx
@@ -36,9 +36,9 @@ queue.task("slow", handler, { timeoutMs: 30_000 });
 Every attempt's error is recorded:
 
 ```ts
-queue.getJobErrors(id); // one entry per failed attempt
-queue.listJobs({ status: "failed" });
-queue.deadLetters(); // exhausted jobs
+await queue.getJobErrors(id); // one entry per failed attempt
+await queue.listJobs({ status: "failed" });
+await queue.deadLetters(); // exhausted jobs
 ```
 
 ## Reacting to outcomes
diff --git a/docs/content/docs/node/more/examples/benchmark.mdx b/docs/content/docs/node/more/examples/benchmark.mdx
index 891d95f6..e4367cd6 100644
--- a/docs/content/docs/node/more/examples/benchmark.mdx
+++ b/docs/content/docs/node/more/examples/benchmark.mdx
@@ -27,14 +27,14 @@ const enqueueMs = Date.now() - t0;
 // 2. Processing throughput — drain the backlog, polling stats until empty.
 const t1 = Date.now();
 const worker = queue.runWorker({ queues: ["default"], batchSize: 64, channelCapacity: 512 });
-while (queue.stats().complete < N) {
+while ((await queue.stats()).complete < N) {
   await new Promise((r) => setTimeout(r, 20));
 }
 const processMs = Date.now() - t1;
 worker.stop();
 
 // 3. End-to-end latency — created → completed across a sample of jobs.
-const sample = queue.listJobs({ status: "complete", limit: 1_000 });
+const sample = await queue.listJobs({ status: "complete", limit: 1_000 });
 const latencies = sample
   .filter((j) => j.completedAt)
   .map((j) => (j.completedAt as number) - j.createdAt)
@@ -82,6 +82,6 @@ latency:  p50 8ms  p99 42ms
 | Pattern | Where |
 | --- | --- |
 | Batched staging | `queue.enqueueMany` |
-| Drain to a target | poll `queue.stats().complete` |
+| Drain to a target | poll `await queue.stats()` for `.complete` |
 | Worker throughput knobs | `runWorker({ batchSize, channelCapacity })` |
 | Latency percentiles | `listJobs` + `completedAt - createdAt` |
diff --git a/docs/content/docs/node/more/examples/bulk-emails.mdx b/docs/content/docs/node/more/examples/bulk-emails.mdx
index 46bf0e4c..293af852 100644
--- a/docs/content/docs/node/more/examples/bulk-emails.mdx
+++ b/docs/content/docs/node/more/examples/bulk-emails.mdx
@@ -70,9 +70,9 @@ export function sendCampaign(recipients: string[], subject: string, body: string
 Watch the backlog drain and catch addresses that exhausted their retries.
 
 ```ts
-console.log(queue.statsByQueue("default")); // { pending, running, complete, failed, dead, cancelled }
+console.log(await queue.statsByQueue("default")); // { pending, running, complete, failed, dead, cancelled }
 
-for (const dead of queue.deadLetters(50)) {
+for (const dead of await queue.deadLetters(50)) {
   console.warn(`gave up on ${dead.taskName}: ${dead.error}`);
 }
 ```
diff --git a/docs/content/docs/node/more/examples/data-pipeline.mdx b/docs/content/docs/node/more/examples/data-pipeline.mdx
index a852ab52..f530b6f9 100644
--- a/docs/content/docs/node/more/examples/data-pipeline.mdx
+++ b/docs/content/docs/node/more/examples/data-pipeline.mdx
@@ -83,7 +83,7 @@ failure history of any step's underlying job.
 ```ts
 import { queue } from "./pipeline";
 
-export function report(runId: string) {
+export async function report(runId: string) {
   const analysis = queue.workflows.analyze(runId);
   if (!analysis) return;
 
@@ -93,7 +93,7 @@ export function report(runId: string) {
   // Inspect the failure history of a specific step.
   const node = analysis.node("enrich");
   if (node?.jobId) {
-    for (const err of queue.getJobErrors(node.jobId)) {
+    for (const err of await queue.getJobErrors(node.jobId)) {
       console.log(`enrich attempt ${err.attempt}: ${err.error}`);
     }
   }
diff --git a/docs/content/docs/node/more/examples/notifications.mdx b/docs/content/docs/node/more/examples/notifications.mdx
index 27a97107..3f51c540 100644
--- a/docs/content/docs/node/more/examples/notifications.mdx
+++ b/docs/content/docs/node/more/examples/notifications.mdx
@@ -74,7 +74,7 @@ export function cancelScheduled(jobId: string) {
 }
 
 // 5. Inspection — what's still pending?
-export function pendingSends() {
+export async function pendingSends() {
   return queue.listJobs({ status: "pending", task: "sendEmail", limit: 50 });
 }
 ```
diff --git a/docs/content/docs/node/more/examples/web-scraper.mdx b/docs/content/docs/node/more/examples/web-scraper.mdx
index 6afef97b..1575e538 100644
--- a/docs/content/docs/node/more/examples/web-scraper.mdx
+++ b/docs/content/docs/node/more/examples/web-scraper.mdx
@@ -56,7 +56,7 @@ queue.task("aggregate", (pages: string[]) => {
 });
 
 queue.task("cleanupCache", async (olderThanMs: number) => {
-  queue.purgeCompleted(olderThanMs);
+  await queue.purgeCompleted(olderThanMs);
 });
 
 // Middleware: one log line per execution and per failure.
diff --git a/sdks/node/README.md b/sdks/node/README.md
index 989beb32..ee49ef37 100644
--- a/sdks/node/README.md
+++ b/sdks/node/README.md
@@ -64,6 +64,12 @@ automatically. Postgres isolates its tables in the `schema` (default
 `"taskito"`); Redis isolates its keys under `prefix`. Override either to share
 or separate state.
 
+Scan-heavy methods (`stats*`, `listJobs`, `deadLetters*`, `purge*`,
+`listWorkers`, `getMetrics`, `getJobErrors`) are async and run off the event
+loop. Single-row operations (`enqueue`, `getJob`, `cancelJob`, …) are sync and
+block for one storage round-trip — negligible on SQLite, but on Postgres/Redis
+that is a network hop, so keep them off latency-critical request paths.
+
 ## Enqueue options
 
 `priority`, `maxRetries`, `timeoutMs`, `delayMs` (delayed run), `uniqueKey`
@@ -92,18 +98,18 @@ queue.requestCancel(jobId); // aborts the task's signal
 ## Inspection & management
 
 ```ts
-queue.stats();              // { pending, running, completed, failed, dead, cancelled }
-queue.statsByQueue("default");
-queue.statsAllQueues();
-queue.listJobs({ status: "failed", limit: 50 });
-queue.getJobErrors(id);
-queue.getMetrics(3600_000, "add");
-
-queue.deadLetters();        // dead-letter entries
+await queue.stats();        // { pending, running, completed, failed, dead, cancelled }
+await queue.statsByQueue("default");
+await queue.statsAllQueues();
+await queue.listJobs({ status: "failed", limit: 50 });
+await queue.getJobErrors(id);
+await queue.getMetrics(3600_000, "add");
+
+await queue.deadLetters();  // dead-letter entries
 queue.retryDead(deadId);    // re-enqueue
 queue.deleteDead(deadId);
-queue.purgeDead(olderThanMs);
-queue.purgeCompleted(olderThanMs);
+await queue.purgeDead(olderThanMs);
+await queue.purgeCompleted(olderThanMs);
 
 queue.pauseQueue("default");
 queue.resumeQueue("default");
diff --git a/sdks/node/src/cli/commands/dlq.ts b/sdks/node/src/cli/commands/dlq.ts
index 24a2d7be..7abad8cd 100644
--- a/sdks/node/src/cli/commands/dlq.ts
+++ b/sdks/node/src/cli/commands/dlq.ts
@@ -11,10 +11,10 @@ export function registerDlq(program: Command): void {
     .description("List dead-letter entries")
     .option("--limit <n>", "page size", "50")
     .option("--offset <n>", "page offset", "0")
-    .action((options: { limit?: string; offset?: string }, command: Command) => {
+    .action(async (options: { limit?: string; offset?: string }, command: Command) => {
       const globals = command.optsWithGlobals() as GlobalOptions;
       const queue = connect(globals);
-      const dead = queue.deadLetters(
+      const dead = await queue.deadLetters(
         nonNegativeIntFlag(options.limit, "limit"),
         nonNegativeIntFlag(options.offset, "offset"),
       );
@@ -52,9 +52,9 @@ export function registerDlq(program: Command): void {
     .command("purge")
     .description("Purge dead-letter entries older than a cutoff")
     .option("--older-than-ms <n>", "age cutoff in ms", "0")
-    .action((options: { olderThanMs?: string }, command: Command) => {
+    .action(async (options: { olderThanMs?: string }, command: Command) => {
       const queue = connect(command.optsWithGlobals() as GlobalOptions);
       const olderThanMs = nonNegativeIntFlag(options.olderThanMs, "older-than-ms") ?? 0;
-      printJson({ purged: queue.purgeDead(olderThanMs) });
+      printJson({ purged: await queue.purgeDead(olderThanMs) });
     });
 }
diff --git a/sdks/node/src/cli/commands/jobs.ts b/sdks/node/src/cli/commands/jobs.ts
index f87dfbed..a3e5caab 100644
--- a/sdks/node/src/cli/commands/jobs.ts
+++ b/sdks/node/src/cli/commands/jobs.ts
@@ -20,10 +20,10 @@ export function registerJobs(program: Command): void {
     .option("-t, --task <t>", "filter by task name")
     .option("--limit <n>", "page size", "50")
     .option("--offset <n>", "page offset", "0")
-    .action((options: JobsOptions, command: Command) => {
+    .action(async (options: JobsOptions, command: Command) => {
       const globals = command.optsWithGlobals() as GlobalOptions;
       const queue = connect(globals);
-      const jobs = queue.listJobs({
+      const jobs = await queue.listJobs({
         status: options.status,
         queue: options.queue,
         task: options.task,
diff --git a/sdks/node/src/cli/commands/stats.ts b/sdks/node/src/cli/commands/stats.ts
index 70f1bf59..6a719b47 100644
--- a/sdks/node/src/cli/commands/stats.ts
+++ b/sdks/node/src/cli/commands/stats.ts
@@ -7,10 +7,10 @@ export function registerStats(program: Command): void {
     .command("stats")
     .description("Show job counts by status")
     .option("-q, --queue <name>", "limit to a single queue")
-    .action((options: { queue?: string }, command: Command) => {
+    .action(async (options: { queue?: string }, command: Command) => {
       const globals = command.optsWithGlobals() as GlobalOptions;
       const queue = connect(globals);
-      const stats = options.queue ? queue.statsByQueue(options.queue) : queue.stats();
+      const stats = options.queue ? await queue.statsByQueue(options.queue) : await queue.stats();
       if (globals.json) {
         printJson(stats);
       } else {
diff --git a/sdks/node/src/contrib/nest.ts b/sdks/node/src/contrib/nest.ts
index 478c883f..02641e39 100644
--- a/sdks/node/src/contrib/nest.ts
+++ b/sdks/node/src/contrib/nest.ts
@@ -39,7 +39,7 @@ export class TaskitoService {
   }
 
   /** Aggregate counts across all queues. */
-  stats(): Stats {
+  stats(): Promise<Stats> {
     return this.queue.stats();
   }
 
@@ -49,7 +49,7 @@ export class TaskitoService {
   }
 
   /** List dead-letter entries. */
-  deadLetters(limit?: number, offset?: number): DeadJob[] {
+  deadLetters(limit?: number, offset?: number): Promise<DeadJob[]> {
     return this.queue.deadLetters(limit, offset);
   }
 }
diff --git a/sdks/node/src/contrib/prometheus.ts b/sdks/node/src/contrib/prometheus.ts
index 370bc058..2d07d03d 100644
--- a/sdks/node/src/contrib/prometheus.ts
+++ b/sdks/node/src/contrib/prometheus.ts
@@ -8,6 +8,9 @@
 import { Counter, register as defaultRegister, Gauge, Histogram, type Registry } from "prom-client";
 import type { Middleware } from "../middleware";
 import type { Queue } from "../queue";
+import { createLogger } from "../utils";
+
+const log = createLogger("prometheus");
 
 /** The per-(registry, namespace) metric bundle, created once and reused. */
 interface TaskitoMetrics {
@@ -170,8 +173,13 @@ export class PrometheusStatsCollector {
     if (this.timer) {
       return;
     }
-    this.sample();
-    this.timer = setInterval(() => this.sample(), this.intervalMs);
+    const sampleSafely = () =>
+      void this.sample().catch((error) => {
+        // A transient storage failure must not kill the polling loop.
+        log.warn(() => "stats sample failed", error);
+      });
+    sampleSafely();
+    this.timer = setInterval(sampleSafely, this.intervalMs);
     this.timer.unref();
   }
 
@@ -184,10 +192,10 @@ export class PrometheusStatsCollector {
   }
 
   /** Read current queue/DLQ stats into the gauges. Public for one-shot scrapes. */
-  sample(): void {
-    for (const [name, stats] of Object.entries(this.queue.statsAllQueues())) {
+  async sample(): Promise<void> {
+    for (const [name, stats] of Object.entries(await this.queue.statsAllQueues())) {
       this.metrics.queueDepth.set({ queue: name }, stats.pending);
     }
-    this.metrics.dlqSize.set(this.queue.stats().dead);
+    this.metrics.dlqSize.set((await this.queue.stats()).dead);
   }
 }
diff --git a/sdks/node/src/contrib/rest.ts b/sdks/node/src/contrib/rest.ts
index 2eb80945..69d03741 100644
--- a/sdks/node/src/contrib/rest.ts
+++ b/sdks/node/src/contrib/rest.ts
@@ -99,14 +99,16 @@ function defineRoutes(resultTimeoutMs: number): RestRoute[] {
       name: "stats",
       method: "GET",
       path: "/stats",
-      handle: (queue) => ({ status: 200, body: queue.stats() }),
+      handle: async (queue) => ({ status: 200, body: await queue.stats() }),
     },
     {
       name: "queue-stats",
       method: "GET",
       path: "/stats/queues",
-      handle: (queue, { query }) => {
-        const body = query.queue ? queue.statsByQueue(query.queue) : queue.statsAllQueues();
+      handle: async (queue, { query }) => {
+        const body = query.queue
+          ? await queue.statsByQueue(query.queue)
+          : await queue.statsAllQueues();
         return { status: 200, body };
       },
     },
@@ -125,7 +127,10 @@ function defineRoutes(resultTimeoutMs: number): RestRoute[] {
       name: "job-errors",
       method: "GET",
       path: "/jobs/:id/errors",
-      handle: (queue, { params }) => ({ status: 200, body: queue.getJobErrors(params.id ?? "") }),
+      handle: async (queue, { params }) => ({
+        status: 200,
+        body: await queue.getJobErrors(params.id ?? ""),
+      }),
     },
     {
       name: "job-result",
@@ -133,7 +138,12 @@ function defineRoutes(resultTimeoutMs: number): RestRoute[] {
       path: "/jobs/:id/result",
       handle: async (queue, { params, query }) => {
         const id = params.id ?? "";
-        const timeoutMs = toInt(query.timeoutMs) ?? resultTimeoutMs;
+        // Clamp client-supplied timeouts so a huge value can't pin the
+        // connection (and its polling loop) open indefinitely.
+        const timeoutMs = Math.min(
+          Math.max(toInt(query.timeoutMs) ?? resultTimeoutMs, 0),
+          resultTimeoutMs,
+        );
         try {
           const result = await queue.result(id, { timeoutMs });
           return { status: 200, body: { jobId: id, status: "completed", result } };
@@ -157,9 +167,9 @@ function defineRoutes(resultTimeoutMs: number): RestRoute[] {
       name: "dead-letters",
       method: "GET",
       path: "/dead-letters",
-      handle: (queue, { query }) => ({
+      handle: async (queue, { query }) => ({
         status: 200,
-        body: queue.deadLetters(toInt(query.limit), toInt(query.offset)),
+        body: await queue.deadLetters(toInt(query.limit), toInt(query.offset)),
       }),
     },
     {
diff --git a/sdks/node/src/dashboard/api.ts b/sdks/node/src/dashboard/api.ts
index ced544bd..a6fc1dea 100644
--- a/sdks/node/src/dashboard/api.ts
+++ b/sdks/node/src/dashboard/api.ts
@@ -12,12 +12,12 @@ function numParam(query: URLSearchParams, key: string): number | undefined {
   return Number.isFinite(num) && num >= 0 ? num : undefined;
 }
 
-export function getStats(queue: Queue) {
-  return { overall: queue.stats(), queues: queue.statsAllQueues() };
+export async function getStats(queue: Queue) {
+  return { overall: await queue.stats(), queues: await queue.statsAllQueues() };
 }
 
-export function getQueues(queue: Queue) {
-  return { paused: queue.listPausedQueues(), stats: queue.statsAllQueues() };
+export async function getQueues(queue: Queue) {
+  return { paused: queue.listPausedQueues(), stats: await queue.statsAllQueues() };
 }
 
 export function getJobs(queue: Queue, query: URLSearchParams) {
diff --git a/sdks/node/src/dashboard/contract.ts b/sdks/node/src/dashboard/contract.ts
index c5521c4d..b8d4e2f7 100644
--- a/sdks/node/src/dashboard/contract.ts
+++ b/sdks/node/src/dashboard/contract.ts
@@ -2,7 +2,7 @@
 // (`dashboard/`) expects. Timestamps are Unix milliseconds throughout.
 
 import type { DeadJob, Job, WorkerInfo } from "../types";
-import type { Webhook } from "../webhooks";
+import type { Delivery, Webhook } from "../webhooks";
 import type { WorkflowNode, WorkflowRun } from "../workflows";
 
 /** Replace header values with a mask so outbound credentials aren't exposed. */
@@ -31,6 +31,26 @@ export function webhookToContract(webhook: Webhook) {
   };
 }
 
+/** Map a webhook delivery to the SPA's `WebhookDelivery` contract. */
+export function deliveryToContract(delivery: Delivery) {
+  return {
+    id: delivery.id,
+    subscription_id: delivery.webhookId,
+    event: delivery.event,
+    payload: delivery.payload,
+    task_name: delivery.taskName,
+    job_id: delivery.jobId,
+    status: delivery.status,
+    attempts: delivery.attempts,
+    response_code: delivery.responseCode,
+    response_body: delivery.responseBody,
+    latency_ms: delivery.latencyMs,
+    error: delivery.error ?? null,
+    created_at: delivery.createdAt,
+    completed_at: delivery.completedAt,
+  };
+}
+
 export function jobToContract(job: Job) {
   return {
     id: job.id,
diff --git a/sdks/node/src/dashboard/handlers.ts b/sdks/node/src/dashboard/handlers.ts
index c5f9acdf..60c17f54 100644
--- a/sdks/node/src/dashboard/handlers.ts
+++ b/sdks/node/src/dashboard/handlers.ts
@@ -8,6 +8,7 @@ import type { WebhookInput } from "../webhooks";
 import type { WorkflowNode } from "../workflows";
 import {
   deadToContract,
+  deliveryToContract,
   jobToContract,
   webhookToContract,
   workerToContract,
@@ -39,16 +40,16 @@ export function stats(queue: Queue) {
   return queue.stats();
 }
 
-export function statsQueues(queue: Queue, url: URL) {
+export async function statsQueues(queue: Queue, url: URL) {
   const name = url.searchParams.get("queue");
-  return name ? { [name]: queue.statsByQueue(name) } : queue.statsAllQueues();
+  return name ? { [name]: await queue.statsByQueue(name) } : queue.statsAllQueues();
 }
 
 export function queuesPaused(queue: Queue) {
   return queue.listPausedQueues();
 }
 
-export function jobs(queue: Queue, url: URL) {
+export async function jobs(queue: Queue, url: URL) {
   const sp = url.searchParams;
   const page = sp.get("page");
   const pageSize = sp.get("pageSize");
@@ -56,15 +57,14 @@ export function jobs(queue: Queue, url: URL) {
   const offset =
     sp.get("offset") ??
     (page !== null && pageSize !== null ? String(Number(page) * Number(pageSize)) : undefined);
-  return queue
-    .listJobs({
-      status: sp.get("status") ?? undefined,
-      queue: sp.get("queue") ?? undefined,
-      task: sp.get("task") ?? undefined,
-      limit: toNonNegative(limit),
-      offset: toNonNegative(offset),
-    })
-    .map(jobToContract);
+  const found = await queue.listJobs({
+    status: sp.get("status") ?? undefined,
+    queue: sp.get("queue") ?? undefined,
+    task: sp.get("task") ?? undefined,
+    limit: toNonNegative(limit),
+    offset: toNonNegative(offset),
+  });
+  return found.map(jobToContract);
 }
 
 export function job(queue: Queue, id: string) {
@@ -72,24 +72,28 @@ export function job(queue: Queue, id: string) {
   return found ? jobToContract(found) : undefined;
 }
 
-export function deadLetters(queue: Queue, url: URL) {
-  return queue.deadLetters(num(url, "limit"), num(url, "offset")).map(deadToContract);
+export async function deadLetters(queue: Queue, url: URL) {
+  const dead = await queue.deadLetters(num(url, "limit"), num(url, "offset"));
+  return dead.map(deadToContract);
 }
 
-export function metrics(queue: Queue, url: URL) {
+export async function metrics(queue: Queue, url: URL) {
   const since = positiveOr(url.searchParams.get("since"), 3600);
   const task = url.searchParams.get("task") ?? undefined;
-  return aggregateByTask(queue.getMetrics(Date.now() - since * 1000, task));
+  return aggregateByTask(await queue.getMetrics(Date.now() - since * 1000, task));
 }
 
-export function timeseries(queue: Queue, url: URL) {
+export async function timeseries(queue: Queue, url: URL) {
   const since = positiveOr(url.searchParams.get("since"), 3600);
   const bucket = positiveOr(url.searchParams.get("bucket"), 60);
-  return bucketTimeseries(queue.getMetrics(Date.now() - since * 1000, undefined), bucket * 1000);
+  return bucketTimeseries(
+    await queue.getMetrics(Date.now() - since * 1000, undefined),
+    bucket * 1000,
+  );
 }
 
-export function workers(queue: Queue) {
-  return queue.listWorkers().map(workerToContract);
+export async function workers(queue: Queue) {
+  return (await queue.listWorkers()).map(workerToContract);
 }
 
 export function eventTypes(queue: Queue) {
@@ -186,20 +190,24 @@ export function deleteWebhook(queue: Queue, id: string) {
 
 export async function testWebhook(queue: Queue, id: string) {
   const delivery = await queue.webhooks.test(id);
-  return delivery ? { status: delivery.status, delivered: delivery.ok } : undefined;
-}
-
-export function webhookDeliveries(queue: Queue, id: string) {
-  return queue.webhooks.deliveries(id).map((delivery) => ({
-    id: delivery.id,
-    subscription_id: delivery.webhookId,
-    event: delivery.event,
-    status: delivery.status,
-    ok: delivery.ok,
-    attempts: delivery.attempts,
-    error: delivery.error ?? null,
-    at: delivery.at,
-  }));
+  // The SPA's TestWebhookResult wants the HTTP status code (or null).
+  return delivery ? { status: delivery.responseCode, delivered: delivery.ok } : undefined;
+}
+
+export function webhookDeliveries(queue: Queue, id: string, url: URL) {
+  const limit = num(url, "limit") ?? 50;
+  const offset = num(url, "offset") ?? 0;
+  const statusFilter = url.searchParams.get("status");
+  const all = queue.webhooks
+    .deliveries(id)
+    .filter((delivery) => !statusFilter || delivery.status === statusFilter)
+    .reverse(); // newest first, matching the Python delivery store
+  return {
+    items: all.slice(offset, offset + limit).map(deliveryToContract),
+    total: all.length,
+    limit,
+    offset,
+  };
 }
 
 /** Parse a snake_case webhook request body into a {@link WebhookInput}. */
diff --git a/sdks/node/src/dashboard/index.ts b/sdks/node/src/dashboard/index.ts
index 522464c8..7cbb8c28 100644
--- a/sdks/node/src/dashboard/index.ts
+++ b/sdks/node/src/dashboard/index.ts
@@ -1,9 +1,12 @@
 import type { Server } from "node:http";
 import { fileURLToPath } from "node:url";
 import type { Queue } from "../index";
+import { createLogger } from "../utils";
 import type { DashboardAuth } from "./auth";
 import { createDashboardServer } from "./server";
 
+const log = createLogger("dashboard");
+
 export interface DashboardOptions {
   /** Port to listen on (default 8787). */
   port?: number;
@@ -33,6 +36,10 @@ export function serveDashboard(queue: Queue, options: DashboardOptions = {}): Se
     options.staticDir ?? defaultStaticDir(),
     options.auth,
   );
+  // A bind failure (e.g. EADDRINUSE) without an 'error' listener crashes the process.
+  server.on("error", (error) => {
+    log.error(() => "dashboard server error", error);
+  });
   server.listen(options.port ?? 8787, options.host ?? "127.0.0.1");
   return server;
 }
diff --git a/sdks/node/src/dashboard/routes.ts b/sdks/node/src/dashboard/routes.ts
index 48262758..113f3cc7 100644
--- a/sdks/node/src/dashboard/routes.ts
+++ b/sdks/node/src/dashboard/routes.ts
@@ -81,7 +81,7 @@ export const routes: Route[] = [
   {
     method: "GET",
     pattern: /^\/api\/webhooks\/([^/]+)\/deliveries$/,
-    handle: (q, _url, p) => h.webhookDeliveries(q, id(p)),
+    handle: (q, url, p) => h.webhookDeliveries(q, id(p), url),
   },
   {
     method: "GET",
diff --git a/sdks/node/src/events.ts b/sdks/node/src/events.ts
index 9824112f..afd468af 100644
--- a/sdks/node/src/events.ts
+++ b/sdks/node/src/events.ts
@@ -34,7 +34,11 @@ export class Emitter {
     }
     for (const handler of set) {
       try {
-        handler(payload);
+        // Promise.resolve captures async listeners' rejections (else they
+        // crash the process as unhandled rejections).
+        void Promise.resolve(handler(payload)).catch(() => {
+          // A listener rejecting must not break other listeners or the worker.
+        });
       } catch {
         // A listener throwing must not break other listeners or the worker.
       }
diff --git a/sdks/node/src/queue.ts b/sdks/node/src/queue.ts
index 0bc9cd73..29e5679d 100644
--- a/sdks/node/src/queue.ts
+++ b/sdks/node/src/queue.ts
@@ -52,6 +52,7 @@ import type {
 import { WebhookManager } from "./webhooks";
 import { Worker } from "./worker";
 import { WorkflowManager } from "./workflows";
+import { WorkflowTracker } from "./workflows/tracker";
 
 /** Construction options for a {@link Queue}. */
 export interface QueueOptions {
@@ -90,6 +91,8 @@ export class Queue<TTasks extends TaskMap = TaskMap> {
   private readonly webhookManager: WebhookManager;
   /** Built lazily — its constructor throws on addons lacking the `workflows` feature. */
   private workflowManager?: WorkflowManager;
+  /** Shared by workers and `workflows.resolveGate()` so gate timers clear. */
+  private workflowTracker?: WorkflowTracker;
 
   constructor(options: QueueOptions = {}) {
     this.native = JsQueue.open(toOpenOptions(options));
@@ -105,11 +108,24 @@ export class Queue<TTasks extends TaskMap = TaskMap> {
   /** Workflow definitions and runs — DAG/linear orchestration over the queue. */
   get workflows(): WorkflowManager {
     if (!this.workflowManager) {
-      this.workflowManager = new WorkflowManager(this.native, this.serializer);
+      this.workflowManager = new WorkflowManager(
+        this.native,
+        this.serializer,
+        this.trackerIfSupported(),
+      );
     }
     return this.workflowManager;
   }
 
+  /** The shared workflow tracker, or `undefined` on addons without workflows. */
+  private trackerIfSupported(): WorkflowTracker | undefined {
+    if (typeof this.native.markWorkflowNodeResult !== "function") {
+      return undefined;
+    }
+    this.workflowTracker ??= new WorkflowTracker(this.native, this.serializer);
+    return this.workflowTracker;
+  }
+
   /** Create a distributed lock handle (not yet acquired). */
   lock(name: string, options?: LockOptions): Lock {
     return new Lock(this.native, name, options);
@@ -388,12 +404,16 @@ export class Queue<TTasks extends TaskMap = TaskMap> {
     const timeoutMs = options?.timeoutMs ?? 60_000;
     const pollMs = options?.pollMs ?? 200;
     const deadline = Date.now() + timeoutMs;
-    const seen = new Set<string>();
+    // Cursor-based: each poll fetches only rows after the last seen log id
+    // (UUIDv7 → time-ordered), instead of rescanning the full history.
+    let cursor: string | undefined;
     for (;;) {
-      yield* this.newPartials(id, seen);
+      const batch = this.newPartials(id, cursor);
+      cursor = batch.cursor;
+      yield* batch.values;
       const job = this.native.getJob(id);
       if (job && TERMINAL_STATUSES.has(job.status)) {
-        yield* this.newPartials(id, seen); // final drain for values committed at completion
+        yield* this.newPartials(id, cursor).values; // drain values committed at completion
         return;
       }
       if (Date.now() >= deadline) {
@@ -408,59 +428,59 @@ export class Queue<TTasks extends TaskMap = TaskMap> {
     return this.native.getTaskLogs(id);
   }
 
-  /** Yield not-yet-seen partial-result entries (dedup by id, robust to same-ms writes). */
-  private *newPartials(id: string, seen: Set<string>): Generator<unknown> {
-    for (const log of this.native.getTaskLogs(id)) {
-      if (log.level !== STREAM_LEVEL || seen.has(log.id)) {
-        continue;
-      }
-      seen.add(log.id);
-      yield decodePartial(log.extra);
-    }
+  /** Partial-result values logged after `cursor`, plus the advanced cursor. */
+  private newPartials(id: string, cursor?: string): { values: unknown[]; cursor?: string } {
+    const logs = this.native.getTaskLogsAfter(id, cursor);
+    return {
+      values: logs
+        .filter((log) => log.level === STREAM_LEVEL)
+        .map((log) => decodePartial(log.extra)),
+      cursor: logs[logs.length - 1]?.id ?? cursor,
+    };
   }
 
   /** Job counts by status across all queues. */
-  stats(): Stats {
+  stats(): Promise<Stats> {
     return this.native.stats();
   }
 
   /** Job counts by status for a single queue. */
-  statsByQueue(queue: string): Stats {
+  statsByQueue(queue: string): Promise<Stats> {
     return this.native.statsByQueue(queue);
   }
 
   /** Job counts by status, keyed by queue name. */
-  statsAllQueues(): Record<string, Stats> {
+  statsAllQueues(): Promise<Record<string, Stats>> {
     return this.native.statsAllQueues();
   }
 
   /** List jobs, optionally filtered and paginated. */
-  listJobs(filter?: JobFilter): Job[] {
+  listJobs(filter?: JobFilter): Promise<Job[]> {
     return this.native.listJobs(filter);
   }
 
   /** Error history for a job (one entry per failed attempt). */
-  getJobErrors(id: string): JobError[] {
+  getJobErrors(id: string): Promise<JobError[]> {
     return this.native.getJobErrors(id);
   }
 
-  /** Per-execution task metrics within the last `sinceMs` milliseconds. */
-  getMetrics(sinceMs: number, task?: string): Metric[] {
+  /** Per-execution task metrics recorded at or after `sinceMs` (Unix epoch ms). */
+  getMetrics(sinceMs: number, task?: string): Promise<Metric[]> {
     return this.native.getMetrics(task ?? null, sinceMs);
   }
 
   /** List dead-letter entries (paginated). */
-  deadLetters(limit?: number, offset?: number): DeadJob[] {
+  deadLetters(limit?: number, offset?: number): Promise<DeadJob[]> {
     return this.native.deadLetters(limit, offset);
   }
 
   /** List dead-letter entries for a single task (paginated, newest first). */
-  deadLettersByTask(taskName: string, limit?: number, offset?: number): DeadJob[] {
+  deadLettersByTask(taskName: string, limit?: number, offset?: number): Promise<DeadJob[]> {
     return this.native.deadLettersByTask(taskName, limit, offset);
   }
 
   /** Delete every dead-letter entry for a task. Returns the count removed. */
-  purgeDeadByTask(taskName: string): number {
+  purgeDeadByTask(taskName: string): Promise<number> {
     return this.native.purgeDeadByTask(taskName);
   }
 
@@ -475,12 +495,12 @@ export class Queue<TTasks extends TaskMap = TaskMap> {
   }
 
   /** Purge dead-letter entries older than `olderThanMs`. Returns the count removed. */
-  purgeDead(olderThanMs: number): number {
+  purgeDead(olderThanMs: number): Promise<number> {
     return this.native.purgeDead(olderThanMs);
   }
 
   /** Purge completed jobs older than `olderThanMs`. Returns the count removed. */
-  purgeCompleted(olderThanMs: number): number {
+  purgeCompleted(olderThanMs: number): Promise<number> {
     return this.native.purgeCompleted(olderThanMs);
   }
 
@@ -500,7 +520,7 @@ export class Queue<TTasks extends TaskMap = TaskMap> {
   }
 
   /** Registered workers (heartbeat + identity). */
-  listWorkers(): WorkerInfo[] {
+  listWorkers(): Promise<WorkerInfo[]> {
     return this.native.listWorkers();
   }
 
@@ -513,6 +533,7 @@ export class Queue<TTasks extends TaskMap = TaskMap> {
       middleware: this.middleware,
       emitter: this.emitter,
       resources: this.resources,
+      workflowTracker: this.trackerIfSupported(),
       run: options,
     });
   }
diff --git a/sdks/node/src/resources/runtime.ts b/sdks/node/src/resources/runtime.ts
index d9e6d858..2251a977 100644
--- a/sdks/node/src/resources/runtime.ts
+++ b/sdks/node/src/resources/runtime.ts
@@ -41,6 +41,9 @@ export class ResourceRuntime {
   /** Register (or replace) a resource definition. */
   register<T>(name: string, definition: ResourceDefinition<T>): void {
     this.defs.set(name, definition as ResourceDefinition);
+    // Drop any built worker instance so "replace" takes effect; the old
+    // instance is still disposed by its queued teardown.
+    this.workerCache.delete(name);
   }
 
   /** True when nothing is registered — lets the worker skip resource wiring. */
diff --git a/sdks/node/src/scaler/server.ts b/sdks/node/src/scaler/server.ts
index 908cc1f9..895ba097 100644
--- a/sdks/node/src/scaler/server.ts
+++ b/sdks/node/src/scaler/server.ts
@@ -44,16 +44,18 @@ export function serveScaler(queue: Queue, options: ScalerOptions = {}): Server {
     }
     if (url.pathname === "/api/scaler" && req.method === "GET") {
       const queueName = url.searchParams.get("queue") ?? defaultQueue;
-      try {
-        const metricValue = queueName
-          ? queue.statsByQueue(queueName).pending
-          : queue.stats().pending;
-        sendJson(res, 200, { metricValue, targetValue, queueName: queueName ?? "*" });
-      } catch (error) {
+      void (async () => {
+        const stats = queueName ? await queue.statsByQueue(queueName) : await queue.stats();
+        sendJson(res, 200, {
+          metricValue: stats.pending,
+          targetValue,
+          queueName: queueName ?? "*",
+        });
+      })().catch((error) => {
         // Keep backend/path details in the logs; never echo them to the caller.
         log.error(() => "scaler metric read failed", error);
         sendJson(res, 500, { error: "internal server error" });
-      }
+      });
       return;
     }
     sendJson(res, 404, { error: "not found" });
diff --git a/sdks/node/src/webhooks/deliverer.ts b/sdks/node/src/webhooks/deliverer.ts
index 97a6eaf6..fd631998 100644
--- a/sdks/node/src/webhooks/deliverer.ts
+++ b/sdks/node/src/webhooks/deliverer.ts
@@ -5,6 +5,7 @@ import type { Delivery, Webhook } from "./types";
 
 const MAX_RECENT = 100;
 const MAX_BACKOFF_MS = 30_000;
+const MAX_RESPONSE_BODY_CHARS = 2048;
 
 const log = createLogger("webhooks");
 
@@ -17,7 +18,8 @@ export class Deliverer {
   }
 
   async deliver(webhook: Webhook, event: EventName, payload: OutcomeEvent): Promise<Delivery> {
-    const body = JSON.stringify({ event, ...payload });
+    const payloadRecord: Record<string, unknown> = { event, ...payload };
+    const body = JSON.stringify(payloadRecord);
     const headers: Record<string, string> = {
       "content-type": "application/json",
       ...webhook.headers,
@@ -27,8 +29,10 @@ export class Deliverer {
       headers["x-taskito-signature"] = `sha256=${signature}`;
     }
 
+    const createdAt = Date.now();
     let attempts = 0;
-    let status = 0;
+    let responseCode: number | null = null;
+    let responseBody: string | null = null;
     let error: string | undefined;
     while (attempts <= webhook.maxRetries) {
       attempts += 1;
@@ -39,14 +43,16 @@ export class Deliverer {
           body,
           signal: AbortSignal.timeout(webhook.timeoutMs),
         });
-        status = response.status;
+        responseCode = response.status;
+        responseBody = await readBody(response);
         if (response.ok) {
           error = undefined;
           break;
         }
         error = `HTTP ${response.status}`;
       } catch (cause) {
-        status = 0;
+        responseCode = null;
+        responseBody = null;
         error = cause instanceof Error ? cause.message : String(cause);
       }
       if (attempts <= webhook.maxRetries) {
@@ -54,15 +60,25 @@ export class Deliverer {
       }
     }
 
+    const completedAt = Date.now();
+    const ok = responseCode !== null && responseCode >= 200 && responseCode < 300;
     const delivery: Delivery = {
       id: randomUUID(),
       webhookId: webhook.id,
       event,
-      status,
-      ok: status >= 200 && status < 300,
+      // The retry chain is inline, so a non-delivered outcome is `dead`.
+      status: ok ? "delivered" : "dead",
+      ok,
       attempts,
+      payload: payloadRecord,
+      taskName: payload.taskName ?? null,
+      jobId: payload.jobId ?? null,
+      responseCode,
+      responseBody,
+      latencyMs: completedAt - createdAt,
       error,
-      at: Date.now(),
+      createdAt,
+      completedAt,
     };
     this.recent.push(delivery);
     if (this.recent.length > MAX_RECENT) {
@@ -78,6 +94,15 @@ export class Deliverer {
   }
 }
 
+/** Read a response body for the delivery log, truncated; `null` if unreadable. */
+async function readBody(response: Response): Promise<string | null> {
+  try {
+    return (await response.text()).slice(0, MAX_RESPONSE_BODY_CHARS);
+  } catch {
+    return null;
+  }
+}
+
 /** Strip credentials and query string from a URL before logging it. */
 function redactUrl(raw: string): string {
   try {
diff --git a/sdks/node/src/webhooks/errors.ts b/sdks/node/src/webhooks/errors.ts
index 970e4ebf..05d8d5c7 100644
--- a/sdks/node/src/webhooks/errors.ts
+++ b/sdks/node/src/webhooks/errors.ts
@@ -1,5 +1,7 @@
+import { TaskitoError } from "../errors";
+
 /** Thrown when a webhook definition fails validation. Dashboard maps it to 400. */
-export class WebhookValidationError extends Error {
+export class WebhookValidationError extends TaskitoError {
   constructor(message: string) {
     super(message);
     this.name = "WebhookValidationError";
diff --git a/sdks/node/src/webhooks/types.ts b/sdks/node/src/webhooks/types.ts
index 33de9d6c..111b1a49 100644
--- a/sdks/node/src/webhooks/types.ts
+++ b/sdks/node/src/webhooks/types.ts
@@ -33,14 +33,28 @@ export interface Webhook {
   updatedAt: number;
 }
 
+/** Terminal state of a delivery attempt-chain. */
+export type DeliveryStatus = "delivered" | "failed" | "dead";
+
 /** The result of a single delivery attempt-chain. */
 export interface Delivery {
   id: string;
   webhookId: string;
   event: EventName;
-  status: number;
+  /** `delivered` on a 2xx; `dead` once every retry is exhausted. */
+  status: DeliveryStatus;
   ok: boolean;
   attempts: number;
+  /** The JSON body that was POSTed. */
+  payload: Record<string, unknown>;
+  taskName: string | null;
+  jobId: string | null;
+  /** Last HTTP status, or `null` when no request completed (network error). */
+  responseCode: number | null;
+  /** Last response body (truncated), or `null` when unread. */
+  responseBody: string | null;
+  latencyMs: number;
   error?: string;
-  at: number;
+  createdAt: number;
+  completedAt: number;
 }
diff --git a/sdks/node/src/worker.ts b/sdks/node/src/worker.ts
index 4cd15c79..e02a1a3d 100644
--- a/sdks/node/src/worker.ts
+++ b/sdks/node/src/worker.ts
@@ -15,7 +15,7 @@ import { type ResourceRuntime, runWithResolver } from "./resources";
 import type { Serializer } from "./serializers";
 import type { QueueLimits, RegisteredTask, WorkerRunOptions } from "./types";
 import { createLogger } from "./utils";
-import { WorkflowTracker } from "./workflows";
+import type { WorkflowTracker } from "./workflows";
 import { CACHE_TASK } from "./workflows/cache";
 
 const log = createLogger("worker");
@@ -39,6 +39,8 @@ export interface WorkerStartParams {
   middleware: readonly Middleware[];
   emitter: Emitter;
   resources: ResourceRuntime;
+  /** The queue's shared tracker (undefined on addons without workflows). */
+  workflowTracker?: WorkflowTracker;
   run?: WorkerRunOptions;
 }
 
@@ -59,10 +61,7 @@ export class Worker {
     const { tasks, queueLimits, serializer, middleware, emitter, resources, run } = params;
 
     // Advance workflow runs as node-jobs settle, unless disabled or unsupported.
-    const tracker =
-      (run?.advanceWorkflows ?? true) && typeof queue.markWorkflowNodeResult === "function"
-        ? new WorkflowTracker(queue, serializer)
-        : null;
+    const tracker = (run?.advanceWorkflows ?? true) ? (params.workflowTracker ?? null) : null;
 
     const taskCallback = async (invocation: JsTaskInvocation): Promise<Buffer> => {
       // Built-in workflow cache-return: echo the single (cached) arg as the result.
@@ -164,7 +163,10 @@ export class Worker {
       for (const mw of middleware) {
         const hook = mw[mapping.hook] as ((e: OutcomeEvent) => void) | undefined;
         try {
-          hook?.(event);
+          // Promise.resolve captures async hooks' rejections too.
+          void Promise.resolve(hook?.(event)).catch((error) => {
+            log.debug(() => `${mapping.hook} middleware hook rejected for ${outcome.jobId}`, error);
+          });
         } catch (error) {
           // outcome hook errors must not break the worker
           log.debug(() => `${mapping.hook} middleware hook threw for ${outcome.jobId}`, error);
diff --git a/sdks/node/src/workflows/analysis.ts b/sdks/node/src/workflows/analysis.ts
index cbc5ad19..647eb310 100644
--- a/sdks/node/src/workflows/analysis.ts
+++ b/sdks/node/src/workflows/analysis.ts
@@ -31,7 +31,8 @@ export interface WorkflowStats {
   pending: number;
 }
 
-const TERMINAL_FAILURE = new Set(["failed", "compensation_failed"]);
+// Failure-side terminal statuses: failed outright, rolled back, or rollback failed.
+const TERMINAL_FAILURE = new Set(["failed", "compensated", "compensation_failed"]);
 
 /**
  * Structural and status analysis of a workflow run's DAG. Built from the graph
@@ -174,7 +175,8 @@ export class WorkflowAnalysis {
     for (const name of this.nodeNames) {
       const status = this.nodeByName.get(name)?.status ?? "pending";
       byStatus[status] = (byStatus[status] ?? 0) + 1;
-      if (status === "completed") {
+      // A cache hit is a success: the node settled with a (reused) result.
+      if (status === "completed" || status === "cache_hit") {
         completed += 1;
       } else if (TERMINAL_FAILURE.has(status)) {
         failed += 1;
diff --git a/sdks/node/src/workflows/manager.ts b/sdks/node/src/workflows/manager.ts
index 5622823d..6caaf2a6 100644
--- a/sdks/node/src/workflows/manager.ts
+++ b/sdks/node/src/workflows/manager.ts
@@ -41,6 +41,8 @@ export class WorkflowManager {
   constructor(
     private readonly native: NativeQueue,
     private readonly serializer: Serializer,
+    /** The queue's shared tracker, so gate resolution clears the worker's timers. */
+    private readonly tracker?: WorkflowTracker,
   ) {
     if (typeof this.native.submitWorkflow !== "function") {
       throw new WorkflowError("the native addon was built without the 'workflows' feature");
@@ -110,7 +112,9 @@ export class WorkflowManager {
    * Idempotent: resolving an already-resolved gate is a no-op.
    */
   resolveGate(runId: string, nodeName: string, approved: boolean, error?: string): void {
-    new WorkflowTracker(this.native, this.serializer).resolveGate(runId, nodeName, approved, error);
+    // No shared tracker (out-of-process resolution) → a throwaway one is fine.
+    const tracker = this.tracker ?? new WorkflowTracker(this.native, this.serializer);
+    tracker.resolveGate(runId, nodeName, approved, error);
   }
 
   /** Approve a waiting gate (shorthand for `resolveGate(runId, node, true)`). */
diff --git a/sdks/node/src/workflows/tracker.ts b/sdks/node/src/workflows/tracker.ts
index e8149bc5..609a3e5f 100644
--- a/sdks/node/src/workflows/tracker.ts
+++ b/sdks/node/src/workflows/tracker.ts
@@ -122,8 +122,9 @@ export class WorkflowTracker {
       }
       this.advance(outcome.jobId, ref.runId, ref.nodeName, succeeded, outcome.error ?? null);
     } catch (error) {
-      // Workflow bookkeeping must never break the worker loop.
-      log.debug(() => `workflow advance for ${outcome.jobId} failed`, error);
+      // Workflow bookkeeping must never break the worker loop — but a swallowed
+      // failure means a stuck run, so it must be visible at the default level.
+      log.error(() => `workflow advance for ${outcome.jobId} failed`, error);
     }
   }
 
@@ -378,9 +379,9 @@ export class WorkflowTracker {
     const fanInNode = this.fanInNodeFor(parent, plan);
     if (fanInNode) {
       this.createFanInJob(runId, fanInNode, childJobIds, plan);
-    } else {
-      this.evaluateSuccessors(runId, parent, plan);
     }
+    // evaluateSuccessors skips fan-in nodes, so the combiner isn't recreated.
+    this.evaluateSuccessors(runId, parent, plan);
   }
 
   /** Read `itemsFrom`'s array result and expand the fan-out into one child per item. */
@@ -446,10 +447,18 @@ export class WorkflowTracker {
     );
   }
 
-  /** Enqueue/expand/skip any deferred successor whose predecessors are now all terminal. */
+  /** Enqueue/expand/skip any successor whose predecessors are now all terminal. */
   private evaluateSuccessors(runId: string, nodeName: string, plan: RunPlan): void {
     for (const succ of plan.successors.get(nodeName) ?? []) {
-      if (!plan.deferred.has(succ) || !this.allPredecessorsTerminal(runId, succ, plan)) {
+      if (!this.allPredecessorsTerminal(runId, succ, plan)) {
+        continue;
+      }
+      if (!plan.deferred.has(succ)) {
+        // The core's fail-fast cascade is suppressed on managed runs, so a
+        // static successor of a failed predecessor must be skipped here.
+        if (!this.shouldExecute(runId, succ, plan)) {
+          this.skipNode(runId, succ, plan);
+        }
         continue;
       }
       // Evaluate the condition first: a not-taken branch is skipped (and the
@@ -535,7 +544,7 @@ export class WorkflowTracker {
       return;
     }
     this.evaluateSuccessors(runId, node, plan);
-    this.evictIfTerminal(runId, this.native.finalizeRunIfTerminal(runId));
+    this.settle(runId, this.native.finalizeRunIfTerminal(runId));
   }
 
   private clearGateTimer(runId: string, node: string): void {
diff --git a/sdks/node/test/core/batch.test.ts b/sdks/node/test/core/batch.test.ts
index 325cd39d..32594ffe 100644
--- a/sdks/node/test/core/batch.test.ts
+++ b/sdks/node/test/core/batch.test.ts
@@ -25,7 +25,7 @@ async function waitFor(predicate: () => boolean, timeoutMs = 4000): Promise<bool
   return false;
 }
 
-it("enqueueMany inserts a batch and returns ids in input order", () => {
+it("enqueueMany inserts a batch and returns ids in input order", async () => {
   const queue = newQueue();
   queue.task("double", (n: number) => n * 2);
 
@@ -37,7 +37,7 @@ it("enqueueMany inserts a batch and returns ids in input order", () => {
   ]);
   expect(ids).toHaveLength(3);
   expect(new Set(ids).size).toBe(3); // distinct ids
-  expect(queue.stats().pending).toBe(3);
+  expect((await queue.stats()).pending).toBe(3);
 
   // Returned ids must line up with the input rows, not just be present.
   expect(ids.map((id) => queue.getJob(id)?.queue)).toEqual(["q0", "q1", "q2"]);
diff --git a/sdks/node/test/core/inspection.test.ts b/sdks/node/test/core/inspection.test.ts
index db7ba6c8..5086138e 100644
--- a/sdks/node/test/core/inspection.test.ts
+++ b/sdks/node/test/core/inspection.test.ts
@@ -21,12 +21,12 @@ it("reports stats and lists jobs", async () => {
   queue.task("add", (a: number, b: number) => a + b);
   const id = queue.enqueue("add", [1, 2]);
 
-  expect(queue.stats().pending).toBe(1);
-  expect(queue.listJobs({ status: "pending" }).map((job) => job.id)).toContain(id);
+  expect((await queue.stats()).pending).toBe(1);
+  expect((await queue.listJobs({ status: "pending" })).map((job) => job.id)).toContain(id);
 
   worker = queue.runWorker();
   expect(await queue.result(id)).toBe(3);
-  expect(queue.stats().completed).toBe(1);
+  expect((await queue.stats()).completed).toBe(1);
 });
 
 it("result() rejects with JobFailedError on a dead job", async () => {
@@ -77,18 +77,18 @@ it("lists and purges dead-letter entries by task", async () => {
   worker = queue.runWorker();
 
   const deadline = Date.now() + 10000;
-  while (Date.now() < deadline && queue.deadLetters().length < 3) {
+  while (Date.now() < deadline && (await queue.deadLetters()).length < 3) {
     await new Promise((resolve) => setTimeout(resolve, 25));
   }
 
-  expect(queue.deadLettersByTask("alpha")).toHaveLength(2);
-  expect(queue.deadLettersByTask("beta")).toHaveLength(1);
+  await expect(queue.deadLettersByTask("alpha")).resolves.toHaveLength(2);
+  await expect(queue.deadLettersByTask("beta")).resolves.toHaveLength(1);
   // Pagination applies within the task's own entries.
-  expect(queue.deadLettersByTask("alpha", 1, 1)).toHaveLength(1);
+  await expect(queue.deadLettersByTask("alpha", 1, 1)).resolves.toHaveLength(1);
 
-  expect(queue.purgeDeadByTask("alpha")).toBe(2);
-  expect(queue.deadLettersByTask("alpha")).toHaveLength(0);
-  expect(queue.deadLetters()).toHaveLength(1);
+  await expect(queue.purgeDeadByTask("alpha")).resolves.toBe(2);
+  await expect(queue.deadLettersByTask("alpha")).resolves.toHaveLength(0);
+  await expect(queue.deadLetters()).resolves.toHaveLength(1);
 });
 
 it("pauses and resumes a queue", () => {
@@ -103,7 +103,7 @@ it("pauses and resumes a queue", () => {
 async function waitForDead(queue: Queue, timeoutMs = 10000): Promise<DeadJob[]> {
   const deadline = Date.now() + timeoutMs;
   while (Date.now() < deadline) {
-    const dead = queue.deadLetters();
+    const dead = await queue.deadLetters();
     if (dead.length > 0) {
       return dead;
     }
diff --git a/sdks/node/test/core/predicates.test.ts b/sdks/node/test/core/predicates.test.ts
index b846bcb2..fac705db 100644
--- a/sdks/node/test/core/predicates.test.ts
+++ b/sdks/node/test/core/predicates.test.ts
@@ -14,11 +14,11 @@ it("enqueues when the gate passes", () => {
   expect(typeof queue.enqueue("charge", [5])).toBe("string");
 });
 
-it("rejects the enqueue when the gate fails", () => {
+it("rejects the enqueue when the gate fails", async () => {
   const queue = newQueue().task("charge", (n: number) => n);
   queue.gate("charge", ({ args }) => args[0] > 0);
   expect(() => queue.enqueue("charge", [-1])).toThrow(PredicateRejectedError);
-  expect(queue.stats().pending).toBe(0);
+  expect((await queue.stats()).pending).toBe(0);
 });
 
 it("requires every gate to pass", () => {
diff --git a/sdks/node/test/integrations/express.test.ts b/sdks/node/test/integrations/express.test.ts
index ee84ef37..7aecd270 100644
--- a/sdks/node/test/integrations/express.test.ts
+++ b/sdks/node/test/integrations/express.test.ts
@@ -32,10 +32,13 @@ async function serve(configure: (app: express.Express) => void): Promise<string>
   return `http://127.0.0.1:${port}`;
 }
 
-async function waitFor(predicate: () => boolean, timeoutMs = 4000): Promise<boolean> {
+async function waitFor(
+  predicate: () => boolean | Promise<boolean>,
+  timeoutMs = 4000,
+): Promise<boolean> {
   const deadline = Date.now() + timeoutMs;
   while (Date.now() < deadline) {
-    if (predicate()) {
+    if (await predicate()) {
       return true;
     }
     await new Promise((resolve) => setTimeout(resolve, 20));
@@ -58,7 +61,7 @@ it("enqueues, runs, and reports results over the REST router", async () => {
   expect(typeof jobId).toBe("string");
 
   worker = queue.runWorker();
-  expect(await waitFor(() => queue.stats().completed >= 1)).toBe(true);
+  expect(await waitFor(async () => (await queue.stats()).completed >= 1)).toBe(true);
 
   const result = await (await fetch(`${base}/tasks/jobs/${jobId}/result`)).json();
   expect(result).toMatchObject({ jobId, status: "completed", result: 5 });
diff --git a/sdks/node/test/integrations/fastify.test.ts b/sdks/node/test/integrations/fastify.test.ts
index b0fd0f24..b86681a0 100644
--- a/sdks/node/test/integrations/fastify.test.ts
+++ b/sdks/node/test/integrations/fastify.test.ts
@@ -21,10 +21,13 @@ function newQueue(): Queue {
   return new Queue({ dbPath: join(mkdtempSync(join(tmpdir(), "taskito-fastify-")), "q.db") });
 }
 
-async function waitFor(predicate: () => boolean, timeoutMs = 4000): Promise<boolean> {
+async function waitFor(
+  predicate: () => boolean | Promise<boolean>,
+  timeoutMs = 4000,
+): Promise<boolean> {
   const deadline = Date.now() + timeoutMs;
   while (Date.now() < deadline) {
-    if (predicate()) {
+    if (await predicate()) {
       return true;
     }
     await new Promise((resolve) => setTimeout(resolve, 20));
@@ -48,7 +51,7 @@ it("serves the REST API through a prefixed plugin", async () => {
   const { jobId } = enqueued.json() as { jobId: string };
 
   worker = queue.runWorker();
-  expect(await waitFor(() => queue.stats().completed >= 1)).toBe(true);
+  expect(await waitFor(async () => (await queue.stats()).completed >= 1)).toBe(true);
 
   const result = await app.inject({ method: "GET", url: `/tasks/jobs/${jobId}/result` });
   expect(result.json()).toMatchObject({ jobId, status: "completed", result: 9 });
diff --git a/sdks/node/test/integrations/nest.test.ts b/sdks/node/test/integrations/nest.test.ts
index 560545a2..d69b685c 100644
--- a/sdks/node/test/integrations/nest.test.ts
+++ b/sdks/node/test/integrations/nest.test.ts
@@ -20,10 +20,13 @@ function newQueue(): Queue {
   return new Queue({ dbPath: join(mkdtempSync(join(tmpdir(), "taskito-nest-")), "q.db") });
 }
 
-async function waitFor(predicate: () => boolean, timeoutMs = 4000): Promise<boolean> {
+async function waitFor(
+  predicate: () => boolean | Promise<boolean>,
+  timeoutMs = 4000,
+): Promise<boolean> {
   const deadline = Date.now() + timeoutMs;
   while (Date.now() < deadline) {
-    if (predicate()) {
+    if (await predicate()) {
       return true;
     }
     await new Promise((resolve) => setTimeout(resolve, 20));
@@ -44,8 +47,8 @@ it("injects TaskitoService bound to the queue", async () => {
 
   const id = service.enqueue("add", [6, 7]);
   worker = queue.runWorker();
-  expect(await waitFor(() => queue.stats().completed >= 1)).toBe(true);
+  expect(await waitFor(async () => (await queue.stats()).completed >= 1)).toBe(true);
 
   expect(await service.result(id)).toBe(13);
-  expect(service.stats().completed).toBeGreaterThanOrEqual(1);
+  expect((await service.stats()).completed).toBeGreaterThanOrEqual(1);
 });
diff --git a/sdks/node/test/observability/prometheus.test.ts b/sdks/node/test/observability/prometheus.test.ts
index cdba77d5..5ea339c5 100644
--- a/sdks/node/test/observability/prometheus.test.ts
+++ b/sdks/node/test/observability/prometheus.test.ts
@@ -17,10 +17,13 @@ function newQueue(): Queue {
   return new Queue({ dbPath: join(mkdtempSync(join(tmpdir(), "taskito-prom-")), "q.db") });
 }
 
-async function waitFor(predicate: () => boolean, timeoutMs = 4000): Promise<boolean> {
+async function waitFor(
+  predicate: () => boolean | Promise<boolean>,
+  timeoutMs = 4000,
+): Promise<boolean> {
   const deadline = Date.now() + timeoutMs;
   while (Date.now() < deadline) {
-    if (predicate()) {
+    if (await predicate()) {
       return true;
     }
     await new Promise((resolve) => setTimeout(resolve, 20));
@@ -45,7 +48,11 @@ it("records job counters and a duration histogram", async () => {
   queue.enqueue("boom", []);
   worker = queue.runWorker();
 
-  expect(await waitFor(() => queue.stats().completed >= 1 && queue.stats().dead >= 1)).toBe(true);
+  expect(
+    await waitFor(
+      async () => (await queue.stats()).completed >= 1 && (await queue.stats()).dead >= 1,
+    ),
+  ).toBe(true);
 
   const text = await register.metrics();
   expect(text).toContain('taskito_jobs_total{task="add",status="completed"} 1');
@@ -80,13 +87,14 @@ it("samples queue depth and DLQ size", async () => {
 
   const collector = new PrometheusStatsCollector(queue, { register, intervalMs: 60_000 });
   collector.start();
+  await collector.sample();
   let text = await register.metrics();
   expect(text).toContain('taskito_queue_depth{queue="default"} 2');
 
   // Drain to the DLQ, then re-sample.
   worker = queue.runWorker();
-  expect(await waitFor(() => queue.stats().dead >= 2)).toBe(true);
-  collector.sample();
+  expect(await waitFor(async () => (await queue.stats()).dead >= 2)).toBe(true);
+  await collector.sample();
   text = await register.metrics();
   expect(text).toContain("taskito_dlq_size 2");
   collector.stop();
diff --git a/sdks/node/test/observability/sentry.test.ts b/sdks/node/test/observability/sentry.test.ts
index 4b54f02a..63500111 100644
--- a/sdks/node/test/observability/sentry.test.ts
+++ b/sdks/node/test/observability/sentry.test.ts
@@ -32,10 +32,13 @@ function captureEvents(): ErrorEvent[] {
   return events;
 }
 
-async function waitFor(predicate: () => boolean, timeoutMs = 4000): Promise<boolean> {
+async function waitFor(
+  predicate: () => boolean | Promise<boolean>,
+  timeoutMs = 4000,
+): Promise<boolean> {
   const deadline = Date.now() + timeoutMs;
   while (Date.now() < deadline) {
-    if (predicate()) {
+    if (await predicate()) {
       return true;
     }
     await new Promise((resolve) => setTimeout(resolve, 20));
@@ -76,7 +79,7 @@ it("does not capture successful jobs", async () => {
   queue.enqueue("ok", []);
   worker = queue.runWorker();
 
-  expect(await waitFor(() => queue.stats().completed >= 1)).toBe(true);
+  expect(await waitFor(async () => (await queue.stats()).completed >= 1)).toBe(true);
   await new Promise((resolve) => setTimeout(resolve, 100));
   expect(events).toHaveLength(0);
 });
diff --git a/sdks/node/test/resources/interception.test.ts b/sdks/node/test/resources/interception.test.ts
index cb7657a6..cc759721 100644
--- a/sdks/node/test/resources/interception.test.ts
+++ b/sdks/node/test/resources/interception.test.ts
@@ -60,7 +60,7 @@ it("onEnqueue can rewrite options before they reach the core", () => {
   expect(job?.metadata).toBe("tagged-by-mw");
 });
 
-it("a throwing onEnqueue aborts the enqueue", () => {
+it("a throwing onEnqueue aborts the enqueue", async () => {
   const queue = newQueue();
   queue.use({
     onEnqueue: (ctx) => {
@@ -72,7 +72,7 @@ it("a throwing onEnqueue aborts the enqueue", () => {
   queue.task("charge", (_amount: number) => undefined);
 
   expect(() => queue.enqueue("charge", [-5])).toThrow("amount must be non-negative");
-  expect(queue.stats().pending).toBe(0); // nothing was enqueued
+  expect((await queue.stats()).pending).toBe(0); // nothing was enqueued
 });
 
 it("runs onEnqueue hooks in registration order", () => {
diff --git a/sdks/node/test/resources/resources.test.ts b/sdks/node/test/resources/resources.test.ts
index 9d0c30d6..c9f011fd 100644
--- a/sdks/node/test/resources/resources.test.ts
+++ b/sdks/node/test/resources/resources.test.ts
@@ -46,6 +46,16 @@ describe("ResourceRuntime", () => {
     expect(builds).toBe(1);
   });
 
+  it("re-registering a worker resource replaces the built instance", async () => {
+    const rt = new ResourceRuntime();
+    rt.register("cfg", { scope: "worker", factory: () => ({ version: 1 }) });
+    const scope = rt.createTaskScope();
+    expect(await scope.resolver("cfg")).toEqual({ version: 1 });
+
+    rt.register("cfg", { scope: "worker", factory: () => ({ version: 2 }) });
+    expect(await scope.resolver("cfg")).toEqual({ version: 2 });
+  });
+
   it("builds a task-scoped resource once per scope, fresh across scopes", async () => {
     const rt = new ResourceRuntime();
     let builds = 0;
diff --git a/sdks/node/test/validation.test.ts b/sdks/node/test/validation.test.ts
index f72c55de..804f1fd2 100644
--- a/sdks/node/test/validation.test.ts
+++ b/sdks/node/test/validation.test.ts
@@ -22,22 +22,22 @@ describe("N-API boundary validation", () => {
     expect(() => queue.enqueue("t", [], { delayMs: -1 })).toThrow(/delayMs/);
   });
 
-  it("rejects negative pagination on listJobs", () => {
+  it("rejects negative pagination on listJobs", async () => {
     const queue = newQueue();
-    expect(() => queue.listJobs({ limit: -1 })).toThrow(/limit/);
-    expect(() => queue.listJobs({ offset: -1 })).toThrow(/offset/);
+    await expect(queue.listJobs({ limit: -1 })).rejects.toThrow(/limit/);
+    await expect(queue.listJobs({ offset: -1 })).rejects.toThrow(/offset/);
   });
 
-  it("rejects an unknown status filter rather than returning everything", () => {
+  it("rejects an unknown status filter rather than returning everything", async () => {
     const queue = newQueue();
-    expect(() => queue.listJobs({ status: "nope" })).toThrow(/unknown status/);
+    await expect(queue.listJobs({ status: "nope" })).rejects.toThrow(/unknown status/);
   });
 
-  it("rejects negative dead-letter pagination and purge cutoffs", () => {
+  it("rejects negative dead-letter pagination and purge cutoffs", async () => {
     const queue = newQueue();
-    expect(() => queue.deadLetters(-1)).toThrow(/limit/);
-    expect(() => queue.purgeDead(-1)).toThrow(/olderThanMs/);
-    expect(() => queue.purgeCompleted(-1)).toThrow(/olderThanMs/);
+    await expect(queue.deadLetters(-1)).rejects.toThrow(/limit/);
+    await expect(queue.purgeDead(-1)).rejects.toThrow(/olderThanMs/);
+    await expect(queue.purgeCompleted(-1)).rejects.toThrow(/olderThanMs/);
   });
 
   it("fails fast on a malformed task rate limit instead of disabling it", () => {
diff --git a/sdks/node/test/worker/events.test.ts b/sdks/node/test/worker/events.test.ts
index 3a7e52e9..68b8060a 100644
--- a/sdks/node/test/worker/events.test.ts
+++ b/sdks/node/test/worker/events.test.ts
@@ -55,6 +55,29 @@ it("emits job.completed and runs before/after/onCompleted middleware", async ()
   expect(calls).toContain("completed");
 });
 
+it("survives async listener and hook rejections", async () => {
+  const queue = newQueue();
+  const events: OutcomeEvent[] = [];
+  // Both an async listener and an async outcome hook reject: neither may crash
+  // the process (unhandledRejection) or block later listeners.
+  queue.on("job.completed", async () => {
+    throw new Error("listener boom");
+  });
+  queue.on("job.completed", (event) => events.push(event));
+  queue.use({
+    onCompleted: (async () => {
+      throw new Error("hook boom");
+    }) as unknown as Middleware["onCompleted"],
+  });
+  queue.task("ok", () => 1);
+
+  queue.enqueue("ok");
+  worker = queue.runWorker();
+
+  expect(await waitFor(() => events.length > 0)).toBe(true);
+  expect(events[0]?.taskName).toBe("ok");
+});
+
 it("emits job.dead and runs onDeadLetter for an exhausted task", async () => {
   const queue = newQueue();
   const dead: OutcomeEvent[] = [];
diff --git a/sdks/node/test/workflows/analysis.test.ts b/sdks/node/test/workflows/analysis.test.ts
index a9fc39b7..a99ff1c3 100644
--- a/sdks/node/test/workflows/analysis.test.ts
+++ b/sdks/node/test/workflows/analysis.test.ts
@@ -67,3 +67,22 @@ it("reports node-status stats", () => {
   expect(stats?.completed).toBe(0); // no worker ran
   expect(stats?.failed).toBe(0);
 });
+
+it("counts cache_hit and compensated as terminal, not pending", () => {
+  const graph = {
+    nodes: [{ name: "a" }, { name: "b" }, { name: "c" }],
+    edges: [
+      { from: "a", to: "b" },
+      { from: "b", to: "c" },
+    ],
+  };
+  const nodes = [
+    { runId: "r", nodeName: "a", status: "cache_hit" },
+    { runId: "r", nodeName: "b", status: "compensated" },
+    { runId: "r", nodeName: "c", status: "compensation_failed" },
+  ];
+  const stats = new WorkflowAnalysis(graph, nodes).stats();
+  expect(stats.pending).toBe(0);
+  expect(stats.completed).toBe(1); // cache_hit is a success
+  expect(stats.failed).toBe(2); // compensated + compensation_failed
+});
diff --git a/sdks/node/test/workflows/dag.test.ts b/sdks/node/test/workflows/dag.test.ts
index 2fe4bd19..3514c506 100644
--- a/sdks/node/test/workflows/dag.test.ts
+++ b/sdks/node/test/workflows/dag.test.ts
@@ -68,6 +68,35 @@ it("fails the run and skips downstream steps when a step dead-letters", async ()
   expect(status.c).toBe("skipped");
 });
 
+it("skips a static successor of a failed step in a managed run", async () => {
+  const queue = freshQueue();
+  const ran: string[] = [];
+  queue.task("boom", () => {
+    throw new Error("kaboom");
+  });
+  queue.task("after", () => ran.push("after"));
+  queue.task("recover", () => ran.push("recover"));
+
+  // `recover`'s condition makes the run tracker-managed, which suppresses the
+  // core's fail-fast cascade — the tracker must settle `next` itself.
+  const handle = queue.workflows
+    .define("managed-static-cascade")
+    .step("risky", "boom", { maxRetries: 0 })
+    .step("next", "after", { after: "risky" })
+    .step("recover", "recover", { after: "risky", condition: "on_failure" })
+    .submit();
+
+  worker = queue.runWorker();
+  const run = await handle.wait({ timeoutMs: 8000 });
+
+  expect(run.state).toBe("failed");
+  const status = Object.fromEntries(handle.nodes().map((n) => [n.nodeName, n.status]));
+  expect(status.risky).toBe("failed");
+  expect(status.next).toBe("skipped");
+  expect(status.recover).toBe("completed");
+  expect(ran).toEqual(["recover"]);
+});
+
 it("lists submitted runs and filters by state", async () => {
   const queue = freshQueue();
   queue.task("noop", () => null);
diff --git a/sdks/node/test/workflows/fanout.test.ts b/sdks/node/test/workflows/fanout.test.ts
index a7d27a11..4cede7d9 100644
--- a/sdks/node/test/workflows/fanout.test.ts
+++ b/sdks/node/test/workflows/fanout.test.ts
@@ -142,6 +142,40 @@ it("runs a fan-out without a fan-in collector", async () => {
   expect(nodesByName(handle).process?.fanOutCount).toBe(2);
 });
 
+it("runs a fan-out's sibling successor alongside its fan-in", async () => {
+  const queue = freshQueue();
+  let combined: number[] | undefined;
+  let notified = false;
+
+  queue.task("source", () => [1, 2]);
+  queue.task("double", (n: number) => n * 2);
+  queue.task("sum", (results: number[]) => {
+    combined = results;
+    return results.length;
+  });
+  queue.task("notify", () => {
+    notified = true;
+  });
+
+  const handle = queue.workflows
+    .define("fanout-sibling")
+    .step("source", "source")
+    .fanOut("process", { after: "source", task: "double", itemsFrom: "source" })
+    .fanIn("collect", { after: "process", task: "sum" })
+    .step("notify", "notify", { after: "process", condition: "always" })
+    .submit();
+
+  worker = queue.runWorker({ queues: ["default"] });
+  const run = await handle.wait({ timeoutMs: 10_000 });
+
+  expect(run.state).toBe("completed");
+  expect(notified).toBe(true);
+  expect(combined?.slice().sort((a, b) => a - b)).toEqual([2, 4]);
+  const nodes = nodesByName(handle);
+  expect(nodes.collect?.status).toBe("completed");
+  expect(nodes.notify?.status).toBe("completed");
+});
+
 it("rejects a fan-in whose target is not a fan-out step", () => {
   const queue = freshQueue();
   expect(() =>
diff --git a/sdks/node/test/workflows/gates.test.ts b/sdks/node/test/workflows/gates.test.ts
index 972a70c1..62d4d062 100644
--- a/sdks/node/test/workflows/gates.test.ts
+++ b/sdks/node/test/workflows/gates.test.ts
@@ -88,6 +88,73 @@ it("skips downstream and fails the run when a gate is rejected", async () => {
   expect(nodes.ship?.status).toBe("skipped");
 });
 
+it("starts saga compensation when a gate is rejected", async () => {
+  const queue = freshQueue();
+  const rollbacks: string[] = [];
+
+  queue.task("reserve", () => "reservation");
+  queue.task("ship", () => 1);
+  queue.task("unreserve", (forward: string) => rollbacks.push(`unreserve:${forward}`));
+
+  const handle = queue.workflows
+    .define("gate-reject-saga")
+    .step("reserve", "reserve", { compensate: "unreserve" })
+    .gate("review", { after: "reserve" })
+    .step("ship", "ship", { after: "review" })
+    .submit();
+
+  worker = queue.runWorker({ queues: ["default"] });
+  expect(await waitFor(() => nodesByName(handle).review?.status === "waiting_approval")).toBe(true);
+
+  queue.workflows.rejectGate(handle.runId, "review", "not allowed");
+
+  const run = await handle.wait({ timeoutMs: 8000 });
+  expect(run.state).toBe("compensated");
+  expect(rollbacks).toEqual(["unreserve:reservation"]);
+  expect(nodesByName(handle).reserve?.status).toBe("compensated");
+});
+
+it("resolves the parent node when a gate inside a sub-workflow is approved", async () => {
+  const queue = freshQueue();
+  const ran: string[] = [];
+
+  queue.task("prep", () => ran.push("prep"));
+  queue.task("childTask", () => ran.push("childTask"));
+  queue.task("finish", () => ran.push("finish"));
+
+  const child = queue.workflows
+    .define("child-gated")
+    .step("a", "childTask")
+    .gate("approval", { after: "a" })
+    .build();
+
+  const handle = queue.workflows
+    .define("parent-gated")
+    .step("prep", "prep")
+    .subWorkflow("sub", { after: "prep", workflow: child })
+    .step("finish", "finish", { after: "sub" })
+    .submit();
+
+  worker = queue.runWorker({ queues: ["default"] });
+
+  expect(await waitFor(() => queue.workflows.children(handle.runId).length > 0)).toBe(true);
+  const childId = queue.workflows.children(handle.runId)[0]?.id ?? "";
+  expect(
+    await waitFor(() =>
+      queue.workflows
+        .nodes(childId)
+        .some((n) => n.nodeName === "approval" && n.status === "waiting_approval"),
+    ),
+  ).toBe(true);
+
+  queue.workflows.approveGate(childId, "approval");
+
+  const run = await handle.wait({ timeoutMs: 10_000 });
+  expect(run.state).toBe("completed");
+  expect(nodesByName(handle).sub?.status).toBe("completed");
+  expect(ran).toEqual(["prep", "childTask", "finish"]);
+});
+
 it("auto-resolves a gate on timeout per onTimeout", async () => {
   const queue = freshQueue();
   let shipped = false;
diff --git a/sdks/python/taskito/workflows/tracker/fan_out.py b/sdks/python/taskito/workflows/tracker/fan_out.py
index 1171b8b0..ac1bd668 100644
--- a/sdks/python/taskito/workflows/tracker/fan_out.py
+++ b/sdks/python/taskito/workflows/tracker/fan_out.py
@@ -93,7 +93,8 @@ def expand_fan_out(
             succ_meta = config.step_metadata.get(successor)
             if succ_meta is not None and succ_meta.get("fan_in") is not None:
                 create_fan_in_job(tracker, run_id, successor, succ_meta, [], config)
-                return
+                break
+        # _evaluate_successors never creates fan-in nodes, so no duplicate.
         tracker._evaluate_successors(run_id, fan_out_node, config)
 
 
@@ -118,14 +119,14 @@ def handle_fan_out_child(
         tracker._try_finalize(run_id)
         return
 
-    # Trigger fan-in.
+    # Trigger fan-in, if any.
     for successor in config.successors.get(parent_name, []):
         meta = config.step_metadata.get(successor)
         if meta is not None and meta.get("fan_in") is not None:
             create_fan_in_job(tracker, run_id, successor, meta, child_job_ids, config)
-            return
+            break
 
-    # No fan-in — evaluate deferred successors.
+    # _evaluate_successors never creates fan-in nodes, so no duplicate.
     tracker._evaluate_successors(run_id, parent_name, config)
 
 
diff --git a/sdks/python/taskito/workflows/tracker/tracker.py b/sdks/python/taskito/workflows/tracker/tracker.py
index b2fe8fca..d19cc2f3 100644
--- a/sdks/python/taskito/workflows/tracker/tracker.py
+++ b/sdks/python/taskito/workflows/tracker/tracker.py
@@ -187,10 +187,8 @@ def _on_cancelled(self, _event_type: EventType, payload: dict[str, Any]) -> None
     # ── Successor evaluation ───────────────────────────────────────
 
     def _evaluate_successors(self, run_id: str, completed_node: str, config: _RunConfig) -> None:
-        """Evaluate and create/skip deferred successor nodes."""
+        """Evaluate and create/skip successor nodes."""
         for successor in config.successors.get(completed_node, []):
-            if successor not in config.deferred_nodes:
-                continue
             meta = config.step_metadata.get(successor)
             if meta is None:
                 continue
@@ -198,6 +196,13 @@ def _evaluate_successors(self, run_id: str, completed_node: str, config: _RunCon
             if not dag.all_predecessors_terminal(self, run_id, successor, config):
                 continue
 
+            if successor not in config.deferred_nodes:
+                # The core's fail-fast cascade is suppressed on managed runs, so
+                # a static successor of a failed predecessor is skipped here.
+                if not dag.should_execute(self, run_id, successor, config):
+                    dag.skip_and_propagate(self, run_id, successor, config)
+                continue
+
             # Fan-out/fan-in nodes: only skip them here (expansion is
             # handled by maybe_trigger_fan_out / handle_fan_out_child).
             if meta.get("fan_out") is not None or meta.get("fan_in") is not None:
diff --git a/sdks/python/tests/workflows/test_workflows_conditions.py b/sdks/python/tests/workflows/test_workflows_conditions.py
index 6c484682..af6f1b24 100644
--- a/sdks/python/tests/workflows/test_workflows_conditions.py
+++ b/sdks/python/tests/workflows/test_workflows_conditions.py
@@ -403,3 +403,40 @@ def on_error() -> str:
     assert final.nodes["collect"].status == NodeStatus.SKIPPED
     assert final.nodes["handle_error"].status == NodeStatus.COMPLETED
     assert collected == ["error handled"]
+
+
+def test_static_successor_skipped_in_managed_run(
+    queue: Queue, workflow_worker: WorkflowWorkerFactory
+) -> None:
+    """A plain (static) successor of a failed step is skipped in a managed run."""
+
+    @queue.task(max_retries=0)
+    def risky() -> str:
+        raise RuntimeError("boom")
+
+    ran: list[str] = []
+
+    @queue.task()
+    def follow_up() -> str:
+        ran.append("follow_up")
+        return "next"
+
+    @queue.task()
+    def recover() -> str:
+        ran.append("recover")
+        return "recovered"
+
+    wf = Workflow(name="managed_static_cascade")
+    wf.step("a", risky)
+    wf.step("b", follow_up, after="a")
+    wf.step("c", recover, after="a", condition="on_failure")
+
+    with workflow_worker():
+        run = queue.submit_workflow(wf)
+        final = run.wait(timeout=20)
+
+    assert final.state == WorkflowState.FAILED
+    assert final.nodes["a"].status == NodeStatus.FAILED
+    assert final.nodes["b"].status == NodeStatus.SKIPPED
+    assert final.nodes["c"].status == NodeStatus.COMPLETED
+    assert ran == ["recover"]
diff --git a/sdks/python/tests/workflows/test_workflows_fan_out.py b/sdks/python/tests/workflows/test_workflows_fan_out.py
index 2c9437bc..d3b55aa5 100644
--- a/sdks/python/tests/workflows/test_workflows_fan_out.py
+++ b/sdks/python/tests/workflows/test_workflows_fan_out.py
@@ -365,3 +365,46 @@ def step_c() -> str:
 
     assert final.state == WorkflowState.COMPLETED
     assert order == ["a", "b", "c"]
+
+
+def test_fan_out_sibling_successor_runs(
+    queue: Queue, workflow_worker: WorkflowWorkerFactory
+) -> None:
+    """A fan-out parent's non-fan-in successor runs alongside the fan-in."""
+
+    @queue.task()
+    def source() -> list[int]:
+        return [1, 2]
+
+    @queue.task()
+    def double(x: int) -> int:
+        return x * 2
+
+    collected: list[int] = []
+
+    @queue.task()
+    def aggregate(results: list[int]) -> str:
+        collected.extend(results)
+        return "done"
+
+    ran: list[str] = []
+
+    @queue.task()
+    def notify() -> str:
+        ran.append("notify")
+        return "sent"
+
+    wf = Workflow(name="fan_out_sibling")
+    wf.step("fetch", source)
+    wf.step("process", double, after="fetch", fan_out="each")
+    wf.step("collect", aggregate, after="process", fan_in="all")
+    wf.step("notify", notify, after="process", condition="always")
+
+    with workflow_worker():
+        run = queue.submit_workflow(wf)
+        final = run.wait(timeout=20)
+
+    assert final.state == WorkflowState.COMPLETED
+    assert sorted(collected) == [2, 4]
+    assert ran == ["notify"]
+    assert final.nodes["notify"].status == NodeStatus.COMPLETED