From b24f81fc130d3d1fa28c7ed6d956fc2377dacb68 Mon Sep 17 00:00:00 2001
From: Cheng <git@zcbenz.com>
Date: Tue, 11 Aug 2026 12:44:31 +0900
Subject: [PATCH 1/3] Fix metal error escaping handler

---
 mlx/backend/common/error.h   | 49 ++++++++++++++++++++++++++++++++++++
 mlx/backend/metal/device.cpp | 25 +++++++++---------
 mlx/backend/metal/device.h   |  4 +--
 mlx/backend/metal/event.cpp  | 19 ++++++--------
 mlx/backend/metal/event.h    | 11 ++++----
 5 files changed, 76 insertions(+), 32 deletions(-)
 create mode 100644 mlx/backend/common/error.h

diff --git a/mlx/backend/common/error.h b/mlx/backend/common/error.h
new file mode 100644
index 0000000000..ba1164f192
--- /dev/null
+++ b/mlx/backend/common/error.h
@@ -0,0 +1,49 @@
+// Copyright © 2026 Apple Inc.
+
+#pragma once
+
+#include <atomic>
+#include <memory>
+#include <string>
+
+namespace mlx::core {
+
+class Error {
+ public:
+  // TODO: Use std::atomic<std::shared_ptr> when it gets supported in Xcode.
+  using Message = std::shared_ptr<std::string>;
+
+  void set_message(Message msg) {
+    std::atomic_store(&message_, std::move(msg));
+  }
+
+  bool valid() const {
+    auto msg = std::atomic_load(&message_);
+    return msg.get();
+  }
+
+  // If |ptr| is a valid event, copy and return true.
+  bool store_if_valid(const Error* ptr) {
+    if (ptr && this != ptr) {
+      Message msg = std::atomic_load(&ptr->message_);
+      if (msg) {
+        set_message(std::move(msg));
+        return true;
+      }
+    }
+    return false;
+  }
+
+  // If current error is valid, throw and clear.
+  void check() {
+    auto msg = std::atomic_exchange(&message_, {});
+    if (msg) {
+      throw std::runtime_error(*msg);
+    }
+  }
+
+ private:
+  Message message_;
+};
+
+} // namespace mlx::core
diff --git a/mlx/backend/metal/device.cpp b/mlx/backend/metal/device.cpp
index 65df5c108c..411951e412 100644
--- a/mlx/backend/metal/device.cpp
+++ b/mlx/backend/metal/device.cpp
@@ -531,20 +531,22 @@ void CommandEncoder::commit(std::function<void()> completion) {
         }
         // If any of the waited event has error in it, poison the encoder.
         for (auto& event : wait_events) {
-          if (event->error()) {
-            error_ = event->error();
+          if (error_.store_if_valid(event->error())) {
             break;
           }
         }
         // Set error only when no error happended before, to preserve the
         // earliest error.
-        if (!error_ && cbuf->status() == MTL::CommandBufferStatusError) {
-          error_ = std::make_shared<std::string>(fmt::format(
-              "[METAL] Command buffer execution failed: {}.",
-              cbuf->error()->localizedDescription()->utf8String()));
+        bool has_error = error_.valid();
+        if (!has_error && cbuf->status() == MTL::CommandBufferStatusError) {
+          error_.set_message(
+              std::make_shared<std::string>(fmt::format(
+                  "[METAL] Command buffer execution failed: {}.",
+                  cbuf->error()->localizedDescription()->utf8String())));
+          has_error = true;
         }
         // Poison all the signaled events when error happened.
-        if (error_) {
+        if (has_error) {
           for (auto& [event, value] : signal_events) {
             event->set_error(error_);
           }
@@ -570,20 +572,17 @@ void CommandEncoder::synchronize() {
   commit();
   cbuf->waitUntilCompleted();
 
-  if (error_ && !exiting_) {
-    auto error = std::move(error_);
-    throw std::runtime_error(*error);
+  if (!exiting_) {
+    error_.check();
   }
 }
 
 MTL::ComputeCommandEncoder* CommandEncoder::get_command_encoder() {
   if (!encoder_) {
+    error_.check();
     encoder_ = NS::RetainPtr(
         buffer_->computeCommandEncoder(MTL::DispatchTypeConcurrent));
     fence_ = NS::TransferPtr(device_.mtl_device()->newFence());
-    // Reset error when user starts to encode new commands, they are supposed to
-    // have handled the error in synchronize() or Event::wait().
-    error_.reset();
   }
   return encoder_.get();
 }
diff --git a/mlx/backend/metal/device.h b/mlx/backend/metal/device.h
index 3bb1e9e3b3..f0c0b9a8e8 100644
--- a/mlx/backend/metal/device.h
+++ b/mlx/backend/metal/device.h
@@ -6,11 +6,11 @@
 #include <functional>
 #include <mutex>
 #include <shared_mutex>
-#include <string>
 #include <unordered_map>
 #include <unordered_set>
 
 #include "mlx/array.h"
+#include "mlx/backend/common/error.h"
 #include "mlx/backend/common/metal_kernel.h"
 #include "mlx/backend/metal/resident.h"
 #include "mlx/device.h"
@@ -123,7 +123,7 @@ class MLX_API CommandEncoder {
   std::vector<std::tuple<std::shared_ptr<EventImpl>, uint64_t>> signal_events_;
 
   // Error from previous commited command buffer.
-  std::shared_ptr<std::string> error_;
+  Error error_;
 
   // Encoder for issuing GPU commands.
   // The members are used within a single ComputeCommandEncoder and will be
diff --git a/mlx/backend/metal/event.cpp b/mlx/backend/metal/event.cpp
index 77f48f0838..0529a72514 100644
--- a/mlx/backend/metal/event.cpp
+++ b/mlx/backend/metal/event.cpp
@@ -26,24 +26,21 @@ EventImpl::~EventImpl() {
 }
 
 void EventImpl::wait(uint64_t value) {
-  check_error();
+  if (auto* p = error(); p) {
+    p->check();
+  }
   mtl_event_->waitUntilSignaledValue(value, -1); // never times out
-  check_error();
+  if (auto* p = error(); p) {
+    p->check();
+  }
 }
 
 void EventImpl::signal(uint64_t value) {
   mtl_event_->setSignaledValue(value);
 }
 
-void EventImpl::set_error(std::shared_ptr<std::string> error) {
-  std::atomic_store(&error_, std::move(error));
-}
-
-void EventImpl::check_error() {
-  auto error = std::atomic_exchange(&error_, {});
-  if (error) {
-    throw std::runtime_error(*error);
-  }
+void EventImpl::set_error(Error& error) {
+  error_.store(&error);
 }
 
 } // namespace metal
diff --git a/mlx/backend/metal/event.h b/mlx/backend/metal/event.h
index c5c82a7cd3..ea0ebdcac0 100644
--- a/mlx/backend/metal/event.h
+++ b/mlx/backend/metal/event.h
@@ -12,11 +12,10 @@ class EventImpl {
 
   void wait(uint64_t value);
   void signal(uint64_t value);
-  void set_error(std::shared_ptr<std::string> error);
-  void check_error();
+  void set_error(Error& error);
 
-  const auto& error() const {
-    return error_;
+  Error* error() const {
+    return error_.load();
   }
 
   auto* mtl_event() {
@@ -24,8 +23,8 @@ class EventImpl {
   }
 
  private:
-  // TODO: Use std::atomic<std::shared_ptr> when it gets supported in Xcode.
-  std::shared_ptr<std::string> error_;
+  // All streams outlive events so pointers would be always valid.
+  std::atomic<Error*> error_;
 
   NS::SharedPtr<MTL::SharedEvent> mtl_event_;
 };

From 3aeae77a0b71425f99553f8114c9e098e5100a44 Mon Sep 17 00:00:00 2001
From: Alessio Pollero <alessio.pollero@gmail.com>
Date: Sun, 21 Jun 2026 15:02:08 +0400
Subject: [PATCH 2/3] Add load tests

---
 python/tests/test_load.py | 28 ++++++++++++++++++++++++++++
 1 file changed, 28 insertions(+)

diff --git a/python/tests/test_load.py b/python/tests/test_load.py
index 1c52f333a6..dc0d1a0b63 100644
--- a/python/tests/test_load.py
+++ b/python/tests/test_load.py
@@ -88,6 +88,34 @@ def test_load_npy_dtype(self):
         with self.assertRaises(Exception):
             out = mx.load(save_file, stream=mx.cpu)
 
+    def test_load_npy_read_error(self):
+        save_file = os.path.join(self.test_dir, "truncated.npy")
+        expected = np.arange(16, dtype=np.float32)
+        np.save(save_file, expected)
+        with open(save_file, "r+b") as f:
+            f.truncate(os.path.getsize(save_file) - expected.nbytes)
+
+        out = mx.load(save_file, stream=mx.cpu)
+        with self.assertRaises(RuntimeError):
+            mx.eval(out)
+
+    def test_async_load_npy_read_error_across_streams(self):
+        save_file = os.path.join(self.test_dir, "truncated_async.npy")
+        expected = np.arange(16, dtype=np.float32)
+        np.save(save_file, expected)
+        with open(save_file, "r+b") as f:
+            f.truncate(os.path.getsize(save_file) - expected.nbytes)
+
+        producer_stream = mx.new_stream(mx.cpu)
+        consumer_stream = mx.new_stream(mx.cpu)
+        out = mx.add(
+            mx.load(save_file, stream=producer_stream),
+            1.0,
+            stream=consumer_stream,
+        )
+        with self.assertRaises(RuntimeError):
+            mx.eval(out)
+
     def test_save_and_load_safetensors(self):
         test_file = os.path.join(self.test_dir, "test.safetensors")
         with self.assertRaises(Exception):

From 9ab2fbb0a2a5adc4edb0fb0b50e631156373854d Mon Sep 17 00:00:00 2001
From: Cheng <git@zcbenz.com>
Date: Mon, 22 Jun 2026 11:04:47 +0900
Subject: [PATCH 3/3] Propagate CPU errors to events

---
 mlx/array.h                      |   1 +
 mlx/backend/common/load.cpp      |   2 +-
 mlx/backend/cpu/encoder.h        |   9 +-
 mlx/backend/cuda/event.cu        | 190 +++++++++++++++----------------
 mlx/backend/cuda/event.h         |  20 +++-
 mlx/backend/cuda/fence.cpp       |  20 ++--
 mlx/backend/metal/device.cpp     |  20 ++--
 mlx/backend/metal/device.h       |  10 +-
 mlx/backend/metal/event.cpp      |  34 +++---
 mlx/backend/metal/event.h        |   7 +-
 mlx/backend/metal/fence.cpp      |   3 +-
 mlx/backend/no_gpu/event.cpp     |  48 ++++----
 mlx/backend/no_gpu/fence.cpp     |  50 +++-----
 mlx/{backend/common => }/error.h |   0
 mlx/event.h                      |  32 ++++++
 mlx/fence.h                      |   7 +-
 mlx/scheduler.cpp                | 116 +++++++++++++++++--
 mlx/scheduler.h                  |  70 ++++--------
 python/tests/test_load.py        |   7 ++
 19 files changed, 369 insertions(+), 277 deletions(-)
 rename mlx/{backend/common => }/error.h (100%)

diff --git a/mlx/array.h b/mlx/array.h
index 8e14ca4726..3f45e9cb9d 100644
--- a/mlx/array.h
+++ b/mlx/array.h
@@ -426,6 +426,7 @@ class MLX_API array {
   }
 
   void detach_event() const {
+    array_desc_->event.check_error();
     array_desc_->event = Event{};
   }
 
diff --git a/mlx/backend/common/load.cpp b/mlx/backend/common/load.cpp
index ce41963de7..b53c92483c 100644
--- a/mlx/backend/common/load.cpp
+++ b/mlx/backend/common/load.cpp
@@ -51,7 +51,7 @@ void Load::eval_cpu(const std::vector<array>& inputs, array& out) {
     }
   };
   auto fut = io::thread_pool().enqueue(std::move(read_task)).share();
-  scheduler::enqueue(stream(), [fut = std::move(fut)]() { fut.wait(); });
+  scheduler::enqueue(stream(), [fut = std::move(fut)]() { fut.get(); });
 }
 
 } // namespace mlx::core
diff --git a/mlx/backend/cpu/encoder.h b/mlx/backend/cpu/encoder.h
index cd015623f6..eb45d64ca0 100644
--- a/mlx/backend/cpu/encoder.h
+++ b/mlx/backend/cpu/encoder.h
@@ -46,11 +46,10 @@ struct MLX_API CommandEncoder {
     auto task = std::bind(std::forward<F>(f), std::forward<Args>(args)...);
     if (num_ops_ == 0) {
       scheduler::notify_new_task(stream_);
-      auto task_wrap = [s = stream_, task = std::move(task)]() mutable {
-        task();
-        scheduler::notify_task_completion(s);
-      };
-      scheduler::enqueue(stream_, std::move(task_wrap));
+      scheduler::enqueue(stream_, std::move(task));
+      // Notify completion separately as |task| may throw exception.
+      scheduler::enqueue(
+          stream_, [s = stream_] { scheduler::notify_task_completion(s); });
     } else {
       scheduler::enqueue(stream_, std::move(task));
     }
diff --git a/mlx/backend/cuda/event.cu b/mlx/backend/cuda/event.cu
index b73937ec38..d3b6f97f5d 100644
--- a/mlx/backend/cuda/event.cu
+++ b/mlx/backend/cuda/event.cu
@@ -113,10 +113,7 @@ void CudaEvent::init_pool() {
   cuda_event_pool();
 }
 
-// Wraps CudaEvent with a few features:
-// 1. The class can be copied.
-// 2. Make wait/record work with CPU streams.
-// 3. Add checks for waiting on un-recorded event.
+// Wraps CudaEvent so it can be copied.
 class CopyableCudaEvent {
  public:
   explicit CopyableCudaEvent(Device& d)
@@ -126,32 +123,24 @@ class CopyableCudaEvent {
                 cudaEventDisableTiming | cudaEventBlockingSync)) {}
 
   void wait() {
+    check_recorded();
     event_->wait();
   }
 
   void wait(Stream s) {
-    if (s.device == mlx::core::Device::cpu) {
-      scheduler::enqueue(s, [*this]() mutable {
-        check_recorded();
-        event_->wait();
-      });
-    } else {
-      check_recorded();
-      auto& encoder = cu::get_command_encoder(s);
-      encoder.commit();
-      event_->wait(encoder.stream());
-    }
+    assert(s.device == mlx::core::Device::gpu);
+    check_recorded();
+    auto& encoder = cu::get_command_encoder(s);
+    encoder.commit();
+    event_->wait(encoder.stream());
   }
 
   void record(Stream s) {
-    if (s.device == mlx::core::Device::cpu) {
-      throw std::runtime_error("CudaEvent can not wait on CPU stream.");
-    } else {
-      auto& encoder = cu::get_command_encoder(s);
-      encoder.commit();
-      event_->record(encoder.stream());
-      recorded_ = true;
-    }
+    assert(s.device == mlx::core::Device::gpu);
+    auto& encoder = cu::get_command_encoder(s);
+    encoder.commit();
+    event_->record(encoder.stream());
+    recorded_ = true;
   }
 
   bool is_signaled() const {
@@ -213,6 +202,11 @@ auto check_gpu_coherency() {
   return coherency;
 }
 
+const CudaStream& signal_stream() {
+  static CudaStream stream(device(0));
+  return stream;
+}
+
 AtomicEvent::AtomicEvent(Device& d) {
   void* buf;
   cudaError_t (*cuda_free)(void*);
@@ -264,14 +258,11 @@ void AtomicEvent::wait(cudaStream_t stream, uint32_t value) {
 
 void AtomicEvent::wait(Stream s, uint32_t value) {
   nvtx3::scoped_range r("cu::AtomicEvent::wait(s)");
-  if (s.device == mlx::core::Device::cpu) {
-    scheduler::enqueue(s, [*this, value]() mutable { wait(value); });
-  } else {
-    auto& encoder = get_command_encoder(s);
-    encoder.commit();
-    wait(encoder.stream(), value);
-    encoder.add_completed_handler([buf = buf_]() {});
-  }
+  assert(s.device == mlx::core::Device::gpu);
+  auto& encoder = get_command_encoder(s);
+  encoder.commit();
+  wait(encoder.stream(), value);
+  encoder.add_completed_handler([buf = buf_]() {});
 }
 
 void AtomicEvent::signal(uint32_t value) {
@@ -289,17 +280,11 @@ void AtomicEvent::signal(cudaStream_t stream, uint32_t value) {
 
 void AtomicEvent::signal(Stream s, uint32_t value) {
   nvtx3::scoped_range r("cu::AtomicEvent::signal(s)");
-  if (s.device == mlx::core::Device::cpu) {
-    // Signal through a GPU stream so the atomic is updated in GPU - updating
-    // the atomic in CPU sometimes does not get GPU notified.
-    scheduler::enqueue(
-        s, [*this, value]() mutable { signal(signal_stream(), value); });
-  } else {
-    auto& encoder = get_command_encoder(s);
-    encoder.commit();
-    signal(encoder.stream(), value);
-    encoder.add_completed_handler([buf = buf_]() {});
-  }
+  assert(s.device == mlx::core::Device::gpu);
+  auto& encoder = get_command_encoder(s);
+  encoder.commit();
+  signal(encoder.stream(), value);
+  encoder.add_completed_handler([buf = buf_]() {});
 }
 
 bool AtomicEvent::is_signaled(uint32_t val) const {
@@ -319,9 +304,21 @@ uint32_t AtomicEvent::value() const {
   }
 }
 
-const CudaStream& AtomicEvent::signal_stream() {
-  static CudaStream stream(device(0));
-  return stream;
+///////////////////////////////////////////////////////////////////////////////
+// EventImpl implementations
+///////////////////////////////////////////////////////////////////////////////
+
+void EventImpl::ensure_created(Stream s, uint64_t signal_value) {
+  if (is_created()) {
+    return;
+  }
+  auto& d = cu::device(s.device);
+  if (s.device == mlx::core::Device::cpu || signal_value > 1) {
+    nvtx3::mark("Using slow AtomicEvent");
+    atomic = std::make_unique<cu::AtomicEvent>(d);
+  } else {
+    cuda = std::make_unique<cu::CopyableCudaEvent>(d);
+  }
 }
 
 } // namespace cu
@@ -330,86 +327,85 @@ const CudaStream& AtomicEvent::signal_stream() {
 // Event implementations
 ///////////////////////////////////////////////////////////////////////////////
 
-namespace {
-
-struct EventImpl {
-  // CudaEvent is preferred when possible because it is fast, however we have
-  // to fallback to AtomicEvent in following cases:
-  // 1. the event is used to wait/signal a cpu stream;
-  // 2. signal value other than 1 has been specified.
-  std::unique_ptr<cu::CopyableCudaEvent> cuda;
-  std::unique_ptr<cu::AtomicEvent> atomic;
-
-  bool is_created() const {
-    return cuda || atomic;
-  }
-
-  void ensure_created(Stream s, uint64_t signal_value) {
-    if (is_created()) {
-      return;
-    }
-    auto& d = cu::device(s.device);
-    if (s.device == mlx::core::Device::cpu || signal_value > 1) {
-      nvtx3::mark("Using slow AtomicEvent");
-      atomic = std::make_unique<cu::AtomicEvent>(d);
-    } else {
-      cuda = std::make_unique<cu::CopyableCudaEvent>(d);
-    }
-  }
-};
-
-} // namespace
-
 Event::Event(Stream s) : stream_(s) {
-  event_ = std::shared_ptr<void>(
-      new EventImpl(), [](void* ptr) { delete static_cast<EventImpl*>(ptr); });
+  event_ = std::make_shared<cu::EventImpl>();
 }
 
 void Event::wait() {
-  auto* event = static_cast<EventImpl*>(event_.get());
-  assert(event->is_created());
-  if (event->cuda) {
+  check_error();
+  auto& event = cast<cu::EventImpl>();
+  assert(event.is_created());
+  if (event.cuda) {
     assert(value() == 1);
-    event->cuda->wait();
+    event.cuda->wait();
   } else {
-    event->atomic->wait(value());
+    event.atomic->wait(value());
   }
   CHECK_CUDA_ERROR(cudaPeekAtLastError());
+  check_error();
 }
 
 void Event::wait(Stream s) {
-  auto* event = static_cast<EventImpl*>(event_.get());
-  assert(event->is_created());
-  if (event->cuda) {
+  auto& event = cast<cu::EventImpl>();
+  assert(event.is_created());
+  if (event.cuda) {
     assert(value() == 1);
-    event->cuda->wait(s);
+    if (s.device == mlx::core::Device::cpu) {
+      scheduler::wait_event(s, *this, [value = value()](Event& self) {
+        self.cast<cu::EventImpl>().cuda->wait();
+      });
+    } else {
+      event.cuda->wait(s);
+    }
   } else {
-    event->atomic->wait(s, value());
+    if (s.device == mlx::core::Device::cpu) {
+      scheduler::wait_event(s, *this, [value = value()](Event& self) {
+        self.cast<cu::EventImpl>().atomic->wait(value);
+      });
+    } else {
+      event.atomic->wait(s, value());
+    }
   }
 }
 
 void Event::signal(Stream s) {
-  auto* event = static_cast<EventImpl*>(event_.get());
-  event->ensure_created(s, value());
-  if (event->cuda) {
+  auto& event = cast<cu::EventImpl>();
+  event.ensure_created(s, value());
+  if (event.cuda) {
     assert(value() == 1);
-    event->cuda->record(s);
+    if (s.device == mlx::core::Device::cpu) {
+      throw std::runtime_error("CudaEvent can not wait on CPU stream.");
+    } else {
+      event.cuda->record(s);
+    }
   } else {
-    event->atomic->signal(s, value());
+    if (s.device == mlx::core::Device::cpu) {
+      // Signal through a GPU stream so the atomic is updated in GPU - updating
+      // the atomic in CPU sometimes does not get GPU notified.
+      scheduler::signal_event(s, *this, [value = value()](Event& self) {
+        self.cast<cu::EventImpl>().atomic->signal(cu::signal_stream(), value);
+      });
+    } else {
+      event.atomic->signal(s, value());
+    }
   }
 }
 
 bool Event::is_signaled() const {
-  auto* event = static_cast<EventImpl*>(event_.get());
-  if (!event->is_created()) {
+  auto& event = cast<cu::EventImpl>();
+  if (!event.is_created()) {
     return false;
   }
-  if (event->cuda) {
+  if (event.cuda) {
     assert(value() == 1);
-    return event->cuda->is_signaled();
+    return event.cuda->is_signaled();
   } else {
-    return event->atomic->is_signaled(value());
+    return event.atomic->is_signaled(value());
   }
 }
 
+std::atomic<Error*>& Event::error() {
+  return cast<cu::EventImpl>().error;
+}
+
 } // namespace mlx::core
diff --git a/mlx/backend/cuda/event.h b/mlx/backend/cuda/event.h
index 53afeb0117..fdeb6a0e78 100644
--- a/mlx/backend/cuda/event.h
+++ b/mlx/backend/cuda/event.h
@@ -13,6 +13,7 @@
 
 namespace mlx::core::cu {
 
+class CopyableCudaEvent;
 class Device;
 
 // RAII-managed move-only wrapper of cudaEvent_t.
@@ -66,8 +67,6 @@ class AtomicEvent {
   uint32_t value() const;
 
  private:
-  const CudaStream& signal_stream();
-
   uint32_t* ptr() const {
     return static_cast<uint32_t*>(buf_.get());
   }
@@ -76,4 +75,21 @@ class AtomicEvent {
   std::shared_ptr<void> buf_;
 };
 
+struct EventImpl {
+  std::atomic<Error*> error;
+
+  // CudaEvent is preferred when possible because it is fast, however we have
+  // to fallback to AtomicEvent in following cases:
+  // 1. the event is used to wait/signal a cpu stream;
+  // 2. signal value other than 1 has been specified.
+  std::unique_ptr<cu::CopyableCudaEvent> cuda;
+  std::unique_ptr<cu::AtomicEvent> atomic;
+
+  bool is_created() const {
+    return cuda || atomic;
+  }
+
+  void ensure_created(Stream s, uint64_t signal_value);
+};
+
 } // namespace mlx::core::cu
diff --git a/mlx/backend/cuda/fence.cpp b/mlx/backend/cuda/fence.cpp
index c6a41f0e60..3a3acdba09 100644
--- a/mlx/backend/cuda/fence.cpp
+++ b/mlx/backend/cuda/fence.cpp
@@ -9,22 +9,23 @@ namespace mlx::core {
 
 struct FenceImpl {
   uint32_t count;
-  cu::AtomicEvent event;
+  Event event;
+
+  FenceImpl(uint32_t count, Stream s) : count(count), event(s) {}
 };
 
 Fence::Fence(Stream s) {
-  fence_ = std::shared_ptr<void>(
-      new FenceImpl{0, cu::device(s.device)},
-      [](void* ptr) { delete static_cast<FenceImpl*>(ptr); });
+  fence_ = std::make_shared<FenceImpl>(0, s);
+  // Ensure that we use AtomicEvent.
+  cast<FenceImpl>().event.cast<cu::EventImpl>().ensure_created(s, 2);
 }
 
 void Fence::wait(Stream s, const array&) {
-  auto* fence = static_cast<FenceImpl*>(fence_.get());
-  fence->event.wait(fence->count);
+  cast<FenceImpl>().event.wait();
 }
 
 void Fence::update(Stream s, const array& a, bool cross_device) {
-  auto* fence = static_cast<FenceImpl*>(fence_.get());
+  auto& f = cast<FenceImpl>();
   if (cross_device) {
     // Move to managed memory if there is a device switch
     auto& cbuf =
@@ -35,8 +36,9 @@ void Fence::update(Stream s, const array& a, bool cross_device) {
       cu::allocator().move_to_unified_memory(cbuf, encoder.stream());
     }
   }
-  fence->count++;
-  fence->event.signal(s, fence->count);
+  f.count++;
+  f.event.set_value(f.count);
+  f.event.signal(s);
 }
 
 } // namespace mlx::core
diff --git a/mlx/backend/metal/device.cpp b/mlx/backend/metal/device.cpp
index 411951e412..2f25f894e4 100644
--- a/mlx/backend/metal/device.cpp
+++ b/mlx/backend/metal/device.cpp
@@ -496,19 +496,15 @@ void CommandEncoder::end_encoding() {
   all_inputs_.clear();
 }
 
-void CommandEncoder::signal_event(
-    std::shared_ptr<EventImpl> event,
-    uint64_t value) {
+void CommandEncoder::signal_event(Event event, uint64_t value) {
   end_encoding();
-  buffer_->encodeSignalEvent(event->mtl_event(), value);
+  buffer_->encodeSignalEvent(event.cast<EventImpl>().mtl_event(), value);
   signal_events_.push_back({std::move(event), value});
 }
 
-void CommandEncoder::wait_event(
-    std::shared_ptr<EventImpl> event,
-    uint64_t value) {
+void CommandEncoder::wait_event(Event event, uint64_t value) {
   end_encoding();
-  buffer_->encodeWait(event->mtl_event(), value);
+  buffer_->encodeWait(event.cast<EventImpl>().mtl_event(), value);
   wait_events_.push_back(std::move(event));
 }
 
@@ -525,13 +521,13 @@ void CommandEncoder::commit(std::function<void()> completion) {
       [&error_ = error_,
        wait_events = std::move(wait_events_),
        signal_events = std::move(signal_events_),
-       completion = std::move(completion)](MTL::CommandBuffer* cbuf) {
+       completion = std::move(completion)](MTL::CommandBuffer* cbuf) mutable {
         if (completion) {
           completion();
         }
         // If any of the waited event has error in it, poison the encoder.
         for (auto& event : wait_events) {
-          if (error_.store_if_valid(event->error())) {
+          if (error_.store_if_valid(event.load_error())) {
             break;
           }
         }
@@ -548,14 +544,14 @@ void CommandEncoder::commit(std::function<void()> completion) {
         // Poison all the signaled events when error happened.
         if (has_error) {
           for (auto& [event, value] : signal_events) {
-            event->set_error(error_);
+            event.set_error(error_);
           }
         }
         // Metal won't signal the events for us on error, manually signal them
         // to avoid infinite waiting.
         if (cbuf->status() == MTL::CommandBufferStatusError) {
           for (auto& [event, value] : signal_events) {
-            event->signal(value);
+            event.cast<EventImpl>().signal(value);
           }
         }
       });
diff --git a/mlx/backend/metal/device.h b/mlx/backend/metal/device.h
index f0c0b9a8e8..2d22283351 100644
--- a/mlx/backend/metal/device.h
+++ b/mlx/backend/metal/device.h
@@ -10,7 +10,6 @@
 #include <unordered_set>
 
 #include "mlx/array.h"
-#include "mlx/backend/common/error.h"
 #include "mlx/backend/common/metal_kernel.h"
 #include "mlx/backend/metal/resident.h"
 #include "mlx/device.h"
@@ -21,7 +20,6 @@ using MTLFCList =
     std::vector<std::tuple<const void*, MTL::DataType, NS::UInteger>>;
 
 class Device;
-class EventImpl;
 
 class MLX_API CommandEncoder {
  public:
@@ -92,8 +90,8 @@ class MLX_API CommandEncoder {
 
   void barrier();
   void end_encoding();
-  void wait_event(std::shared_ptr<EventImpl> event, uint64_t value);
-  void signal_event(std::shared_ptr<EventImpl> event, uint64_t value);
+  void wait_event(Event event, uint64_t value);
+  void signal_event(Event event, uint64_t value);
   bool needs_commit() const;
   void commit(std::function<void()> completion = nullptr);
   void synchronize();
@@ -119,8 +117,8 @@ class MLX_API CommandEncoder {
   uint64_t sets_attached_{0};
 
   // The events hooked to current command buffer.
-  std::vector<std::shared_ptr<EventImpl>> wait_events_;
-  std::vector<std::tuple<std::shared_ptr<EventImpl>, uint64_t>> signal_events_;
+  std::vector<Event> wait_events_;
+  std::vector<std::tuple<Event, uint64_t>> signal_events_;
 
   // Error from previous commited command buffer.
   Error error_;
diff --git a/mlx/backend/metal/event.cpp b/mlx/backend/metal/event.cpp
index 0529a72514..38a387c9c0 100644
--- a/mlx/backend/metal/event.cpp
+++ b/mlx/backend/metal/event.cpp
@@ -26,23 +26,13 @@ EventImpl::~EventImpl() {
 }
 
 void EventImpl::wait(uint64_t value) {
-  if (auto* p = error(); p) {
-    p->check();
-  }
   mtl_event_->waitUntilSignaledValue(value, -1); // never times out
-  if (auto* p = error(); p) {
-    p->check();
-  }
 }
 
 void EventImpl::signal(uint64_t value) {
   mtl_event_->setSignaledValue(value);
 }
 
-void EventImpl::set_error(Error& error) {
-  error_.store(&error);
-}
-
 } // namespace metal
 
 ///////////////////////////////////////////////////////////////////////////////
@@ -54,36 +44,40 @@ Event::Event(Stream stream) : stream_(stream) {
 }
 
 void Event::wait() {
-  static_cast<metal::EventImpl*>(event_.get())->wait(value());
+  check_error();
+  cast<metal::EventImpl>().wait(value());
+  check_error();
 }
 
 void Event::wait(Stream stream) {
-  auto impl = std::static_pointer_cast<metal::EventImpl>(event_);
   if (stream.device == Device::cpu) {
-    scheduler::enqueue(stream, [impl = std::move(impl), value = value()]() {
-      impl->wait(value);
+    scheduler::wait_event(stream, *this, [value = value()](Event& self) {
+      self.cast<metal::EventImpl>().wait(value);
     });
   } else {
     auto& encoder = metal::get_command_encoder(stream);
-    encoder.wait_event(std::move(impl), value());
+    encoder.wait_event(*this, value());
   }
 }
 
 void Event::signal(Stream stream) {
-  auto impl = std::static_pointer_cast<metal::EventImpl>(event_);
   if (stream.device == Device::cpu) {
-    scheduler::enqueue(stream, [impl = std::move(impl), value = value()]() {
-      impl->signal(value);
+    scheduler::signal_event(stream, *this, [value = value()](Event& self) {
+      self.cast<metal::EventImpl>().signal(value);
     });
   } else {
     auto& encoder = metal::get_command_encoder(stream);
-    encoder.signal_event(std::move(impl), value());
+    encoder.signal_event(*this, value());
   }
 }
 
 bool Event::is_signaled() const {
-  auto* mtl_event = static_cast<metal::EventImpl*>(event_.get())->mtl_event();
+  auto* mtl_event = cast<metal::EventImpl>().mtl_event();
   return mtl_event->signaledValue() >= value();
 }
 
+std::atomic<Error*>& Event::error() {
+  return cast<metal::EventImpl>().error();
+}
+
 } // namespace mlx::core
diff --git a/mlx/backend/metal/event.h b/mlx/backend/metal/event.h
index ea0ebdcac0..d1e43fa02f 100644
--- a/mlx/backend/metal/event.h
+++ b/mlx/backend/metal/event.h
@@ -12,13 +12,12 @@ class EventImpl {
 
   void wait(uint64_t value);
   void signal(uint64_t value);
-  void set_error(Error& error);
 
-  Error* error() const {
-    return error_.load();
+  auto& error() {
+    return error_;
   }
 
-  auto* mtl_event() {
+  auto* mtl_event() const {
     return mtl_event_.get();
   }
 
diff --git a/mlx/backend/metal/fence.cpp b/mlx/backend/metal/fence.cpp
index 6fdd57a5f6..70dd0e33bd 100644
--- a/mlx/backend/metal/fence.cpp
+++ b/mlx/backend/metal/fence.cpp
@@ -41,8 +41,7 @@ struct FenceImpl {
 };
 
 Fence::Fence(Stream stream) {
-  auto dtor = [](void* ptr) { delete static_cast<FenceImpl*>(ptr); };
-  fence_ = std::shared_ptr<void>(new FenceImpl(stream), dtor);
+  fence_ = std::make_shared<FenceImpl>(stream);
 }
 
 void Fence::wait(Stream stream, const array& x) {
diff --git a/mlx/backend/no_gpu/event.cpp b/mlx/backend/no_gpu/event.cpp
index 6dde047ab4..8966b77613 100644
--- a/mlx/backend/no_gpu/event.cpp
+++ b/mlx/backend/no_gpu/event.cpp
@@ -12,42 +12,52 @@ struct EventCounter {
   uint64_t value{0};
   std::mutex mtx;
   std::condition_variable cv;
+  std::atomic<Error*> error;
+
+  void wait(uint64_t val) {
+    std::unique_lock<std::mutex> lk(mtx);
+    if (value >= val) {
+      return;
+    }
+    cv.wait(lk, [this, val] { return value >= val; });
+  }
 };
 
 Event::Event(Stream stream) : stream_(stream) {
-  auto dtor = [](void* ptr) { delete static_cast<EventCounter*>(ptr); };
-  event_ = std::shared_ptr<void>(new EventCounter{}, dtor);
+  event_ = std::make_shared<EventCounter>();
 }
 
 void Event::wait() {
-  auto ec = static_cast<EventCounter*>(event_.get());
-  std::unique_lock<std::mutex> lk(ec->mtx);
-  if (ec->value >= value()) {
-    return;
-  }
-  ec->cv.wait(lk, [value = value(), ec] { return ec->value >= value; });
+  check_error();
+  cast<EventCounter>().wait(value());
+  check_error();
 }
 
 void Event::wait(Stream stream) {
-  scheduler::enqueue(stream, [*this]() mutable { wait(); });
+  scheduler::wait_event(stream, *this, [value = value()](Event& self) {
+    self.cast<EventCounter>().wait(value);
+  });
 }
 
 void Event::signal(Stream stream) {
-  scheduler::enqueue(stream, [*this]() mutable {
-    auto ec = static_cast<EventCounter*>(event_.get());
+  scheduler::signal_event(stream, *this, [value = value()](Event& self) {
+    auto& ec = self.cast<EventCounter>();
     {
-      std::lock_guard<std::mutex> lk(ec->mtx);
-      ec->value = value();
+      std::lock_guard lk(ec.mtx);
+      ec.value = value;
     }
-    ec->cv.notify_all();
+    ec.cv.notify_all();
   });
 }
 
 bool Event::is_signaled() const {
-  auto ec = static_cast<EventCounter*>(event_.get());
-  {
-    std::lock_guard<std::mutex> lk(ec->mtx);
-    return (ec->value >= value());
-  }
+  auto& ec = cast<EventCounter>();
+  std::lock_guard lk(ec.mtx);
+  return ec.value >= value();
+}
+
+std::atomic<Error*>& Event::error() {
+  return cast<EventCounter>().error;
 }
+
 } // namespace mlx::core
diff --git a/mlx/backend/no_gpu/fence.cpp b/mlx/backend/no_gpu/fence.cpp
index cd66d23cfe..05852c860b 100644
--- a/mlx/backend/no_gpu/fence.cpp
+++ b/mlx/backend/no_gpu/fence.cpp
@@ -1,54 +1,30 @@
 // Copyright © 2024 Apple Inc.
 
-#include <condition_variable>
-#include <mutex>
-
 #include "mlx/fence.h"
-#include "mlx/scheduler.h"
+#include "mlx/event.h"
 
 namespace mlx::core {
 
 struct FenceImpl {
-  uint32_t count{0};
-  uint32_t value{0};
-  std::mutex mtx;
-  std::condition_variable cv;
+  uint32_t count;
+  Event event;
+
+  FenceImpl(uint32_t count, Stream s) : count(count), event(s) {}
 };
 
-Fence::Fence(Stream) {
-  auto dtor = [](void* ptr) { delete static_cast<FenceImpl*>(ptr); };
-  fence_ = std::shared_ptr<void>(new FenceImpl{}, dtor);
+Fence::Fence(Stream s) {
+  fence_ = std::make_shared<FenceImpl>(0, s);
 }
 
-void Fence::wait(Stream stream, const array&) {
-  auto& f = *static_cast<FenceImpl*>(fence_.get());
-  if (stream.device == Device::cpu) {
-    scheduler::enqueue(stream, [count = f.count, fence_ = fence_]() mutable {
-      auto& f = *static_cast<FenceImpl*>(fence_.get());
-      std::unique_lock<std::mutex> lk(f.mtx);
-      if (f.value >= count) {
-        return;
-      }
-      f.cv.wait(lk, [&f, count] { return f.value >= count; });
-    });
-  } else {
-    throw std::runtime_error("[Fence::wait] Invalid stream.");
-  }
+void Fence::wait(Stream s, const array&) {
+  cast<FenceImpl>().event.wait(s);
 }
 
-void Fence::update(Stream stream, const array&, bool) {
-  auto& f = *static_cast<FenceImpl*>(fence_.get());
+void Fence::update(Stream s, const array&, bool) {
+  auto& f = cast<FenceImpl>();
   f.count++;
-  if (stream.device == Device::cpu) {
-    scheduler::enqueue(stream, [count = f.count, fence_ = fence_]() mutable {
-      auto& f = *static_cast<FenceImpl*>(fence_.get());
-      std::unique_lock<std::mutex> lk(f.mtx);
-      f.value = count;
-      f.cv.notify_all();
-    });
-  } else {
-    throw std::runtime_error("[Fence::update] Invalid stream.");
-  }
+  f.event.set_value(f.count);
+  f.event.signal(s);
 }
 
 } // namespace mlx::core
diff --git a/mlx/backend/common/error.h b/mlx/error.h
similarity index 100%
rename from mlx/backend/common/error.h
rename to mlx/error.h
diff --git a/mlx/event.h b/mlx/event.h
index 66a6a75df5..cf2d5cc7d6 100644
--- a/mlx/event.h
+++ b/mlx/event.h
@@ -5,6 +5,7 @@
 #include <memory>
 #include <stdexcept>
 
+#include "mlx/error.h"
 #include "mlx/stream.h"
 
 namespace mlx::core {
@@ -26,6 +27,26 @@ class Event {
   // Check if the event has been signaled at its current value
   bool is_signaled() const;
 
+  // Associate an error to the event
+  void set_error(Error& err) {
+    error().store(&err);
+  }
+
+  // Get the error associated with the event
+  Error* load_error() const {
+    if (!valid()) {
+      return nullptr;
+    }
+    return error().load();
+  }
+
+  // Throw and clear the associated error
+  void check_error() {
+    if (auto* p = load_error(); p) {
+      p->check();
+    }
+  }
+
   // Check if the event is valid
   bool valid() const {
     return event_ != nullptr;
@@ -47,7 +68,18 @@ class Event {
     return stream_;
   }
 
+  template <typename T>
+  auto& cast() const {
+    return *static_cast<T*>(event_.get());
+  }
+
  private:
+  std::atomic<Error*>& error();
+
+  const std::atomic<Error*>& error() const {
+    return const_cast<Event*>(this)->error();
+  }
+
   // Default constructed stream should never be used
   // since the event is not yet valid
   Stream stream_{0, Device::cpu};
diff --git a/mlx/fence.h b/mlx/fence.h
index 0ececdb6d7..3fd5da333b 100644
--- a/mlx/fence.h
+++ b/mlx/fence.h
@@ -32,8 +32,13 @@ class Fence {
   void update(Stream stream, const array& x, bool cross_device);
   void wait(Stream stream, const array& x);
 
+  template <typename T>
+  auto& cast() const {
+    return *static_cast<T*>(fence_.get());
+  }
+
  private:
-  std::shared_ptr<void> fence_{nullptr};
+  std::shared_ptr<void> fence_;
 };
 
 } // namespace mlx::core
diff --git a/mlx/scheduler.cpp b/mlx/scheduler.cpp
index 7507917f5b..616200072b 100644
--- a/mlx/scheduler.cpp
+++ b/mlx/scheduler.cpp
@@ -1,8 +1,11 @@
 // Copyright © 2023-2026 Apple Inc.
 
-#include "mlx/scheduler.h"
+#include <future>
+#include <thread>
+
 #include "mlx/backend/cpu/eval.h"
 #include "mlx/backend/gpu/eval.h"
+#include "mlx/scheduler.h"
 #include "mlx/utils.h"
 
 namespace mlx::core {
@@ -33,6 +36,58 @@ void clear_streams() {
 
 namespace scheduler {
 
+struct StreamThread {
+  std::mutex mtx;
+  std::queue<std::function<void()>> q;
+  std::condition_variable cond;
+  bool stop;
+  std::thread thread;
+  Error error;
+
+  StreamThread() : stop(false), thread(&StreamThread::thread_fn, this) {}
+
+  ~StreamThread() {
+    {
+      std::lock_guard<std::mutex> lk(mtx);
+      stop = true;
+    }
+    cond.notify_one();
+    thread.join();
+  }
+
+  void thread_fn() {
+    while (true) {
+      std::function<void()> task;
+      {
+        std::unique_lock<std::mutex> lk(mtx);
+        cond.wait(lk, [this] { return !this->q.empty() || this->stop; });
+        if (q.empty() && stop) {
+          return;
+        }
+        task = std::move(q.front());
+        q.pop();
+      }
+
+      task();
+    }
+  }
+
+  void enqueue(std::function<void()> f) {
+    if (is_main_thread()) {
+      error.check();
+    }
+    {
+      std::lock_guard<std::mutex> lk(mtx);
+      if (stop) {
+        throw std::runtime_error(
+            "Cannot enqueue work after stream is stopped.");
+      }
+      q.emplace(std::move(f));
+    }
+    cond.notify_one();
+  }
+};
+
 Scheduler::Scheduler() {
   is_main_thread();
   gpu::init();
@@ -41,23 +96,62 @@ Scheduler::Scheduler() {
 Scheduler::~Scheduler() = default;
 
 void Scheduler::enqueue(Stream s, std::function<void()> task) {
-  StreamThread* st = nullptr;
+  auto& st = get_thread(s);
+  st.enqueue([&st, task = std::move(task)]() mutable {
+    try {
+      task();
+    } catch (const std::exception& error) {
+      // Set error to stream only when no error happended before, to preserve
+      // the earliest error.
+      if (!st.error.valid()) {
+        st.error.set_message(std::make_shared<std::string>(error.what()));
+      }
+    }
+  });
+}
+
+void Scheduler::wait_event(
+    Stream s,
+    Event event,
+    std::function<void(Event&)> task) {
+  assert(s.device == Device::cpu);
+  auto& st = get_thread(s);
+  st.enqueue([&st, event = std::move(event), task = std::move(task)]() mutable {
+    task(event);
+    // Poison current stream if the waited event has error.
+    st.error.store_if_valid(event.load_error());
+  });
+}
+
+void Scheduler::signal_event(
+    Stream s,
+    Event event,
+    std::function<void(Event&)> task) {
+  assert(s.device == Device::cpu);
+  auto& st = get_thread(s);
+  st.enqueue([&st, event = std::move(event), task = std::move(task)]() mutable {
+    // Poison the signal event if current stream has error.
+    if (st.error.valid()) {
+      event.set_error(st.error);
+    }
+    task(event);
+  });
+}
+
+StreamThread& Scheduler::get_thread(Stream s) {
   {
     std::shared_lock lock(threads_mtx_);
     auto it = threads_.find(s.index);
     if (it != threads_.end()) {
-      st = it->second.get();
+      return *it->second.get();
     }
   }
-  if (!st) {
-    std::unique_lock lock(threads_mtx_);
-    auto it = threads_.find(s.index);
-    if (it == threads_.end()) {
-      it = threads_.emplace(s.index, std::make_unique<StreamThread>()).first;
-    }
-    st = it->second.get();
+  std::unique_lock lock(threads_mtx_);
+  auto it = threads_.find(s.index);
+  if (it == threads_.end()) {
+    it = threads_.emplace(s.index, std::make_unique<StreamThread>()).first;
   }
-  st->enqueue(std::move(task));
+  return *it->second.get();
 }
 
 // Leak the scheduler singleton on all platforms. During static destruction,
diff --git a/mlx/scheduler.h b/mlx/scheduler.h
index c84ab62855..7c05b83689 100644
--- a/mlx/scheduler.h
+++ b/mlx/scheduler.h
@@ -3,66 +3,19 @@
 #pragma once
 
 #include <atomic>
-#include <future>
 #include <queue>
 #include <shared_mutex>
-#include <thread>
 #include <unordered_map>
 
 #include "mlx/api.h"
 #include "mlx/backend/gpu/eval.h"
 #include "mlx/device.h"
 #include "mlx/stream.h"
+#include "mlx/utils.h"
 
 namespace mlx::core::scheduler {
 
-struct StreamThread {
-  std::mutex mtx;
-  std::queue<std::function<void()>> q;
-  std::condition_variable cond;
-  bool stop;
-  std::thread thread;
-
-  StreamThread() : stop(false), thread(&StreamThread::thread_fn, this) {}
-
-  ~StreamThread() {
-    {
-      std::lock_guard<std::mutex> lk(mtx);
-      stop = true;
-    }
-    cond.notify_one();
-    thread.join();
-  }
-
-  void thread_fn() {
-    while (true) {
-      std::function<void()> task;
-      {
-        std::unique_lock<std::mutex> lk(mtx);
-        cond.wait(lk, [this] { return !this->q.empty() || this->stop; });
-        if (q.empty() && stop) {
-          return;
-        }
-        task = std::move(q.front());
-        q.pop();
-      }
-
-      task();
-    }
-  }
-
-  void enqueue(std::function<void()> f) {
-    {
-      std::lock_guard<std::mutex> lk(mtx);
-      if (stop) {
-        throw std::runtime_error(
-            "Cannot enqueue work after stream is stopped.");
-      }
-      q.emplace(std::move(f));
-    }
-    cond.notify_one();
-  }
-};
+class StreamThread;
 
 class MLX_API Scheduler {
  public:
@@ -76,6 +29,8 @@ class MLX_API Scheduler {
   Scheduler& operator=(Scheduler&&) = delete;
 
   void enqueue(Stream s, std::function<void()> task);
+  void wait_event(Stream s, Event event, std::function<void(Event&)> task);
+  void signal_event(Stream s, Event event, std::function<void(Event&)> task);
 
   void notify_new_task(const Stream& stream) {
     {
@@ -110,6 +65,8 @@ class MLX_API Scheduler {
  private:
   friend Stream mlx::core::new_stream(Device d);
 
+  StreamThread& get_thread(Stream s);
+
   int n_active_tasks_{0};
   std::unordered_map<int, std::unique_ptr<StreamThread>> threads_;
   std::shared_mutex threads_mtx_;
@@ -120,8 +77,19 @@ class MLX_API Scheduler {
 MLX_API Scheduler& scheduler();
 
 template <typename F>
-void enqueue(const Stream& stream, F&& f) {
-  scheduler().enqueue(stream, std::forward<F>(f));
+inline void enqueue(Stream s, F&& f) {
+  scheduler().enqueue(s, std::forward<F>(f));
+}
+
+// Like enqueue but the task is used for processing the passed event.
+template <typename F>
+inline void wait_event(Stream s, Event event, F&& f) {
+  scheduler().wait_event(s, std::move(event), std::forward<F>(f));
+}
+
+template <typename F>
+inline void signal_event(Stream s, Event event, F&& f) {
+  scheduler().signal_event(s, std::move(event), std::forward<F>(f));
 }
 
 inline int n_active_tasks() {
diff --git a/python/tests/test_load.py b/python/tests/test_load.py
index dc0d1a0b63..9dd9bff838 100644
--- a/python/tests/test_load.py
+++ b/python/tests/test_load.py
@@ -115,6 +115,13 @@ def test_async_load_npy_read_error_across_streams(self):
         )
         with self.assertRaises(RuntimeError):
             mx.eval(out)
+        # The error should propagate on both streams, but the Event impl of
+        # CUDA backend signals via gpu stream which adds a Fence wait which
+        # does a synchronous wait, so error surfaced early in producer_stream
+        # before poisoning the producer_stream.
+        if not mx.cuda.is_available():
+            with self.assertRaises(RuntimeError):
+                mx.synchronize(producer_stream)
 
     def test_save_and_load_safetensors(self):
         test_file = os.path.join(self.test_dir, "test.safetensors")