diff --git a/openpilot/tools/cabana/SConscript b/openpilot/tools/cabana/SConscript
index 387129709..bc366ab6f 100644
--- a/openpilot/tools/cabana/SConscript
+++ b/openpilot/tools/cabana/SConscript
@@ -99,7 +99,7 @@ cabana_env.Command(assets, "assets/assets.qrc", f"rcc $SOURCES -o $TARGET")
 cabana_env.Depends(assets, Glob('/assets/*', exclude=[assets, "assets/assets.o"]))
 
 cabana_srcs = ['mainwin.cc', 'streams/pandastream.cc', 'streams/devicestream.cc', 'streams/livestream.cc', 'streams/abstractstream.cc', 'streams/replaystream.cc', 'binaryview.cc', 'historylog.cc', 'videowidget.cc', 'signalview.cc',
-               'streams/routes.cc', 'dbc/dbc.cc', 'dbc/dbcfile.cc', 'dbc/dbcmanager.cc', 'dbc/dbcqt.cc',
+               'streams/routes.cc', 'dbc/dbc.cc', 'dbc/dbcfile.cc', 'dbc/dbcmanager.cc',
                'utils/export.cc', 'utils/util.cc', 'utils/elidedlabel.cc',
                'chart/chartswidget.cc', 'chart/chart.cc', 'chart/signalselector.cc', 'chart/tiplabel.cc', 'chart/sparkline.cc',
                'commands.cc', 'messageswidget.cc', 'streamselector.cc', 'settings.cc', 'panda.cc',
diff --git a/openpilot/tools/cabana/binaryview.cc b/openpilot/tools/cabana/binaryview.cc
index 5e919dc6a..2f0167107 100644
--- a/openpilot/tools/cabana/binaryview.cc
+++ b/openpilot/tools/cabana/binaryview.cc
@@ -1,5 +1,4 @@
 #include "tools/cabana/binaryview.h"
-#include "tools/cabana/dbc/dbcqt.h"
 
 #include <algorithm>
 
@@ -36,8 +35,8 @@ BinaryView::BinaryView(QWidget *parent) : QTableView(parent) {
   setMouseTracking(true);
   setHorizontalScrollBarPolicy(Qt::ScrollBarAlwaysOff);
 
-  QObject::connect(dbcNotifier(), &QtDBCNotifier::DBCFileChanged, this, &BinaryView::refresh);
-  QObject::connect(undoNotifier(), &QtUndoNotifier::indexChanged, this, &BinaryView::refresh);
+  connections_.push_back(dbc()->fileChanged.connect([this]() { refresh(); }));
+  connections_.push_back(UndoStack::instance()->indexChanged.connect([this]() { refresh(); }));
 
   addShortcuts();
   setWhatsThis(R"(
diff --git a/openpilot/tools/cabana/binaryview.h b/openpilot/tools/cabana/binaryview.h
index c49067a1a..0b0159b42 100644
--- a/openpilot/tools/cabana/binaryview.h
+++ b/openpilot/tools/cabana/binaryview.h
@@ -100,5 +100,6 @@ private:
   bool is_message_active = false;
   const cabana::Signal *resize_sig = nullptr;
   const cabana::Signal *hovered_sig = nullptr;
+  Connections connections_;
   friend class BinaryItemDelegate;
 };
diff --git a/openpilot/tools/cabana/cabana.cc b/openpilot/tools/cabana/cabana.cc
index 8b21143fa..e3a185034 100644
--- a/openpilot/tools/cabana/cabana.cc
+++ b/openpilot/tools/cabana/cabana.cc
@@ -146,19 +146,19 @@ int main(int argc, char *argv[]) {
   AbstractStream *stream = nullptr;
 
   if (args.msgq) {
-    stream = new DeviceStream(&app);
+    stream = new DeviceStream();
   } else if (!args.zmq.empty()) {
-    stream = new DeviceStream(&app, QString::fromStdString(args.zmq));
+    stream = new DeviceStream(QString::fromStdString(args.zmq));
   } else if (args.panda || !args.panda_serial.empty()) {
     try {
-      stream = new PandaStream(&app, {.serial = args.panda_serial});
+      stream = new PandaStream({.serial = args.panda_serial});
     } catch (std::exception &e) {
       fprintf(stderr, "%s\n", e.what());
       return 0;
     }
 #ifdef __linux__
   } else if (SocketCanStream::available() && !args.socketcan.empty()) {
-    stream = new SocketCanStream(&app, {.device = args.socketcan});
+    stream = new SocketCanStream({.device = args.socketcan});
 #endif
   } else {
     uint32_t replay_flags = REPLAY_FLAG_NONE;
@@ -174,7 +174,7 @@ int main(int argc, char *argv[]) {
       route = DEMO_ROUTE;
     }
     if (!route.isEmpty()) {
-      auto replay_stream = std::make_unique<ReplayStream>(&app);
+      auto replay_stream = std::make_unique<ReplayStream>();
       if (!replay_stream->loadRoute(route.toStdString(), args.data_dir, replay_flags, args.auto_source)) {
         return 0;
       }
diff --git a/openpilot/tools/cabana/cameraview.cc b/openpilot/tools/cabana/cameraview.cc
index 9bd6b9be1..b8c70ab1b 100644
--- a/openpilot/tools/cabana/cameraview.cc
+++ b/openpilot/tools/cabana/cameraview.cc
@@ -9,13 +9,11 @@
 #include <QPainter>
 
 #include "common/yuv.h"
+#include "tools/cabana/utils/util.h"
 
 CameraWidget::CameraWidget(std::string stream_name, VisionStreamType type, QWidget* parent) :
                           stream_name(stream_name), active_stream_type(type), requested_stream_type(type), QWidget(parent) {
   setAttribute(Qt::WA_OpaquePaintEvent);
-  qRegisterMetaType<std::set<VisionStreamType>>("availableStreams");
-  QObject::connect(this, &CameraWidget::vipcThreadFrameReceived, this, &CameraWidget::vipcFrameReceived, Qt::QueuedConnection);
-  QObject::connect(this, &CameraWidget::vipcAvailableStreamsUpdated, this, &CameraWidget::availableStreamsUpdated, Qt::QueuedConnection);
   QObject::connect(QApplication::instance(), &QCoreApplication::aboutToQuit, this, &CameraWidget::stopVipcThread);
 }
 
@@ -38,10 +36,6 @@ void CameraWidget::stopVipcThread() {
   }
 }
 
-void CameraWidget::availableStreamsUpdated(std::set<VisionStreamType> streams) {
-  available_streams = streams;
-}
-
 void CameraWidget::paintEvent(QPaintEvent *event) {
   QPainter p(this);
   p.fillRect(rect(), bg);
@@ -67,10 +61,6 @@ void CameraWidget::paintEvent(QPaintEvent *event) {
   p.drawImage(video_rect, rgb_frame);
 }
 
-void CameraWidget::vipcFrameReceived() {
-  update();
-}
-
 void CameraWidget::vipcThread() {
   VisionStreamType cur_stream = requested_stream_type;
   std::unique_ptr<VisionIpcClient> vipc_client;
@@ -93,7 +83,11 @@ void CameraWidget::vipcThread() {
         std::this_thread::sleep_for(std::chrono::milliseconds(100));
         continue;
       }
-      emit vipcAvailableStreamsUpdated(streams);
+      utils::runOnMainThread([this, alive = std::weak_ptr<bool>(alive_), streams]() {
+        if (alive.expired()) return;
+        available_streams = streams;
+        availableStreamsUpdated(streams);
+      });
 
       if (!vipc_client->connect(false)) {
         std::this_thread::sleep_for(std::chrono::milliseconds(100));
@@ -112,7 +106,9 @@ void CameraWidget::vipcThread() {
         std::lock_guard lk(frame_lock);
         rgb_frame.swap(rgb_back);
       }
-      emit vipcThreadFrameReceived();
+      utils::runOnMainThread([this, alive = std::weak_ptr<bool>(alive_)]() {
+        if (!alive.expired()) update();
+      });
     }
   }
 }
diff --git a/openpilot/tools/cabana/cameraview.h b/openpilot/tools/cabana/cameraview.h
index 3b55dd9ed..ac292a701 100644
--- a/openpilot/tools/cabana/cameraview.h
+++ b/openpilot/tools/cabana/cameraview.h
@@ -1,6 +1,7 @@
 #pragma once
 
 #include <atomic>
+#include <memory>
 #include <mutex>
 #include <set>
 #include <string>
@@ -11,11 +12,10 @@
 #include <QWidget>
 
 #include "openpilot/cereal/visionstream.h"
+#include "tools/cabana/core/observable.h"
 #include "msgq/visionipc/visionipc_client.h"
 
 class CameraWidget : public QWidget {
-  Q_OBJECT
-
 public:
   explicit CameraWidget(std::string stream_name, VisionStreamType stream_type, QWidget* parent = nullptr);
   ~CameraWidget();
@@ -23,16 +23,14 @@ public:
   VisionStreamType getStreamType() { return active_stream_type; }
   void stopVipcThread();
 
-signals:
-  void clicked();
-  void vipcThreadFrameReceived();
-  void vipcAvailableStreamsUpdated(std::set<VisionStreamType>);
+  Observable<> clicked;
+  Observable<std::set<VisionStreamType>> availableStreamsUpdated;  // invoked on the main thread
 
 protected:
   void paintEvent(QPaintEvent *event) override;
   void showEvent(QShowEvent *event) override;
   void hideEvent(QHideEvent *event) override { stopVipcThread(); }
-  void mouseReleaseEvent(QMouseEvent *event) override { emit clicked(); }
+  void mouseReleaseEvent(QMouseEvent *event) override { clicked(); }
   void vipcThread();
   void clearFrames();
 
@@ -47,10 +45,5 @@ protected:
   std::thread vipc_thread;
   std::atomic<bool> vipc_exit = false;
   std::mutex frame_lock;
-
-protected slots:
-  void vipcFrameReceived();
-  void availableStreamsUpdated(std::set<VisionStreamType> streams);
+  std::shared_ptr<bool> alive_ = std::make_shared<bool>(true);
 };
-
-Q_DECLARE_METATYPE(std::set<VisionStreamType>);
diff --git a/openpilot/tools/cabana/chart/chart.cc b/openpilot/tools/cabana/chart/chart.cc
index 8496c2699..beece1235 100644
--- a/openpilot/tools/cabana/chart/chart.cc
+++ b/openpilot/tools/cabana/chart/chart.cc
@@ -1,5 +1,4 @@
 #include "tools/cabana/chart/chart.h"
-#include "tools/cabana/dbc/dbcqt.h"
 
 #include <algorithm>
 #include <limits>
@@ -37,10 +36,10 @@ ChartView::ChartView(const std::pair<double, double> &x_range, ChartsWidget *par
   createToolButtons();
   signal_value_font.setPointSize(9);
 
-  QObject::connect(dbcNotifier(), &QtDBCNotifier::signalRemoved, this, &ChartView::signalRemoved);
-  QObject::connect(dbcNotifier(), &QtDBCNotifier::signalUpdated, this, &ChartView::signalUpdated);
-  QObject::connect(dbcNotifier(), &QtDBCNotifier::msgRemoved, this, &ChartView::msgRemoved);
-  QObject::connect(dbcNotifier(), &QtDBCNotifier::msgUpdated, this, &ChartView::msgUpdated);
+  connections_.push_back(dbc()->signalRemoved.connect([this](const cabana::Signal *sig) { signalRemoved(sig); }));
+  connections_.push_back(dbc()->signalUpdated.connect([this](const cabana::Signal *sig) { signalUpdated(sig); }));
+  connections_.push_back(dbc()->msgRemoved.connect([this](MessageId id) { msgRemoved(id); }));
+  connections_.push_back(dbc()->msgUpdated.connect([this](MessageId id) { msgUpdated(id); }));
 }
 
 void ChartView::createToolButtons() {
diff --git a/openpilot/tools/cabana/chart/chart.h b/openpilot/tools/cabana/chart/chart.h
index b03623475..d7227b6cf 100644
--- a/openpilot/tools/cabana/chart/chart.h
+++ b/openpilot/tools/cabana/chart/chart.h
@@ -126,5 +126,6 @@ private:
   double tooltip_x = -1;
   QFont signal_value_font;
   ChartsWidget *charts_widget;
+  Connections connections_;
   friend class ChartsWidget;
 };
diff --git a/openpilot/tools/cabana/chart/chartswidget.cc b/openpilot/tools/cabana/chart/chartswidget.cc
index 3144c5132..8aa095b65 100644
--- a/openpilot/tools/cabana/chart/chartswidget.cc
+++ b/openpilot/tools/cabana/chart/chartswidget.cc
@@ -1,5 +1,4 @@
 #include "tools/cabana/chart/chartswidget.h"
-#include "tools/cabana/dbc/dbcqt.h"
 
 #include <algorithm>
 #include <future>
@@ -76,10 +75,10 @@ ChartsWidget::ChartsWidget(QWidget *parent) : QFrame(parent) {
   redo_zoom_action = toolbar->addAction(utils::icon("arrow-clockwise"), tr("Redo Zoom"), [this]() { zoom_undo_stack.redo(); });
   undo_zoom_action->setEnabled(false);
   redo_zoom_action->setEnabled(false);
-  zoom_undo_stack.setCallbacks({.index_changed = [this]() {
+  connections_.push_back(zoom_undo_stack.indexChanged.connect([this]() {
     undo_zoom_action->setEnabled(zoom_undo_stack.canUndo());
     redo_zoom_action->setEnabled(zoom_undo_stack.canRedo());
-  }});
+  }));
   reset_zoom_action = toolbar->addWidget(reset_zoom_btn = new ToolButton("zoom-out", tr("Reset Zoom")));
   reset_zoom_btn->setToolButtonStyle(Qt::ToolButtonTextBesideIcon);
 
@@ -122,16 +121,16 @@ ChartsWidget::ChartsWidget(QWidget *parent) : QFrame(parent) {
   align_timer->setSingleShot(true);
   QObject::connect(align_timer, &QTimer::timeout, this, &ChartsWidget::alignCharts);
   QObject::connect(auto_scroll_timer, &QTimer::timeout, this, &ChartsWidget::doAutoScroll);
-  QObject::connect(dbcNotifier(), &QtDBCNotifier::DBCFileChanged, this, &ChartsWidget::removeAll);
-  QObject::connect(can, &AbstractStream::eventsMerged, this, &ChartsWidget::eventsMerged);
-  QObject::connect(can, &AbstractStream::msgsReceived, this, &ChartsWidget::updateState);
-  QObject::connect(can, &AbstractStream::seeking, this, &ChartsWidget::updateState);
-  QObject::connect(can, &AbstractStream::timeRangeChanged, this, &ChartsWidget::timeRangeChanged);
+  connections_.push_back(dbc()->fileChanged.connect([this]() { removeAll(); }));
+  connections_.push_back(can->eventsMerged.connect([this](const MessageEventsMap &events) { eventsMerged(events); }));
+  connections_.push_back(can->msgsReceived.connect([this](const std::set<MessageId> *, bool) { updateState(); }));
+  connections_.push_back(can->seeking.connect([this](double) { updateState(); }));
+  connections_.push_back(can->timeRangeChanged.connect([this](const auto &range) { timeRangeChanged(range); }));
   QObject::connect(range_slider, &QSlider::valueChanged, this, &ChartsWidget::setMaxChartRange);
   QObject::connect(new_plot_btn, &QToolButton::clicked, this, &ChartsWidget::newChart);
   QObject::connect(remove_all_btn, &QToolButton::clicked, this, &ChartsWidget::removeAll);
   QObject::connect(reset_zoom_btn, &QToolButton::clicked, this, &ChartsWidget::zoomReset);
-  QObject::connect(&settings, &Settings::changed, this, &ChartsWidget::settingChanged);
+  connections_.push_back(settings.changed.connect([this]() { settingChanged(); }));
   QObject::connect(new_tab_btn, &QToolButton::clicked, this, &ChartsWidget::newTab);
   QObject::connect(this, &ChartsWidget::seriesChanged, this, &ChartsWidget::updateTabBar);
   QObject::connect(tabbar, &QTabBar::tabCloseRequested, this, &ChartsWidget::removeTab);
diff --git a/openpilot/tools/cabana/chart/chartswidget.h b/openpilot/tools/cabana/chart/chartswidget.h
index 8b3003dcd..4c143242b 100644
--- a/openpilot/tools/cabana/chart/chartswidget.h
+++ b/openpilot/tools/cabana/chart/chartswidget.h
@@ -123,6 +123,7 @@ private:
   QTimer *align_timer;
   int current_theme = 0;
   bool value_tip_visible_ = false;
+  Connections connections_;
   friend class ChartView;
   friend class ChartsContainer;
 };
diff --git a/openpilot/tools/cabana/chart/signalselector.cc b/openpilot/tools/cabana/chart/signalselector.cc
index 85832e796..aab38f537 100644
--- a/openpilot/tools/cabana/chart/signalselector.cc
+++ b/openpilot/tools/cabana/chart/signalselector.cc
@@ -1,5 +1,4 @@
 #include "tools/cabana/chart/signalselector.h"
-#include "tools/cabana/dbc/dbcqt.h"
 
 #include <QDialogButtonBox>
 #include <QGridLayout>
diff --git a/openpilot/tools/cabana/commands.cc b/openpilot/tools/cabana/commands.cc
index b47bf90b0..c6cdd9b1b 100644
--- a/openpilot/tools/cabana/commands.cc
+++ b/openpilot/tools/cabana/commands.cc
@@ -28,22 +28,22 @@ void UndoStack::clear() {
   bool was_clean = isClean();
   commands_.clear();
   index_ = clean_index_ = 0;
-  if (callbacks_.index_changed) callbacks_.index_changed();
-  if (!was_clean && callbacks_.clean_changed) callbacks_.clean_changed(true);
+  indexChanged();
+  if (!was_clean) cleanChanged(true);
 }
 
 void UndoStack::setClean() {
   if (!isClean()) {
     clean_index_ = index_;
-    if (callbacks_.clean_changed) callbacks_.clean_changed(true);
+    cleanChanged(true);
   }
 }
 
 void UndoStack::setIndex(int index) {
   bool was_clean = isClean();
   index_ = index;
-  if (callbacks_.index_changed) callbacks_.index_changed();
-  if (isClean() != was_clean && callbacks_.clean_changed) callbacks_.clean_changed(isClean());
+  indexChanged();
+  if (isClean() != was_clean) cleanChanged(isClean());
 }
 
 UndoStack *UndoStack::instance() {
@@ -51,18 +51,6 @@ UndoStack *UndoStack::instance() {
   return &undo_stack;
 }
 
-QtUndoNotifier::QtUndoNotifier(QObject *parent) : QObject(parent) {
-  UndoStack::instance()->setCallbacks({
-    .index_changed = [this]() { emit indexChanged(); },
-    .clean_changed = [this](bool clean) { emit cleanChanged(clean); },
-  });
-}
-
-QtUndoNotifier *undoNotifier() {
-  static QtUndoNotifier notifier;
-  return &notifier;
-}
-
 // EditMsgCommand
 
 EditMsgCommand::EditMsgCommand(const MessageId &id, const std::string &name, int size,
diff --git a/openpilot/tools/cabana/commands.h b/openpilot/tools/cabana/commands.h
index 200a4f2f5..1e4049951 100644
--- a/openpilot/tools/cabana/commands.h
+++ b/openpilot/tools/cabana/commands.h
@@ -1,13 +1,11 @@
 #pragma once
 
-#include <functional>
 #include <memory>
 #include <string>
 #include <utility>
 #include <vector>
 
-#include <QObject>
-
+#include "tools/cabana/core/observable.h"
 #include "tools/cabana/dbc/dbcmanager.h"
 #include "tools/cabana/streams/abstractstream.h"
 
@@ -21,11 +19,6 @@ public:
 
 class UndoStack {
 public:
-  struct Callbacks {
-    std::function<void()> index_changed;
-    std::function<void(bool)> clean_changed;
-  };
-
   void push(UndoCommand *cmd);  // takes ownership and calls redo()
   void undo();
   void redo();
@@ -36,31 +29,18 @@ public:
   bool canRedo() const { return index_ < (int)commands_.size(); }
   std::string undoText() const { return canUndo() ? commands_[index_ - 1]->text : ""; }
   std::string redoText() const { return canRedo() ? commands_[index_]->text : ""; }
-  void setCallbacks(Callbacks callbacks) { callbacks_ = std::move(callbacks); }
   static UndoStack *instance();
 
+  Observable<> indexChanged;
+  Observable<bool> cleanChanged;
+
 private:
   void setIndex(int index);
   std::vector<std::unique_ptr<UndoCommand>> commands_;
   int index_ = 0;
   int clean_index_ = 0;
-  Callbacks callbacks_;
 };
 
-// emits Qt signals for the global undo stack
-class QtUndoNotifier : public QObject {
-  Q_OBJECT
-
-public:
-  explicit QtUndoNotifier(QObject *parent = nullptr);
-
-signals:
-  void indexChanged();
-  void cleanChanged(bool clean);
-};
-
-QtUndoNotifier *undoNotifier();
-
 class EditMsgCommand : public UndoCommand {
 public:
   EditMsgCommand(const MessageId &id, const std::string &name, int size, const std::string &node,
diff --git a/openpilot/tools/cabana/core/observable.h b/openpilot/tools/cabana/core/observable.h
new file mode 100644
index 000000000..0c992bf1d
--- /dev/null
+++ b/openpilot/tools/cabana/core/observable.h
@@ -0,0 +1,83 @@
+#pragma once
+
+#include <functional>
+#include <map>
+#include <memory>
+#include <utility>
+#include <vector>
+
+namespace observable_detail {
+struct HandlerTable {
+  virtual ~HandlerTable() = default;
+  virtual void erase(int id) = 0;
+};
+}  // namespace observable_detail
+
+// disconnects on destruction; safe to outlive the Observable
+class Connection {
+public:
+  Connection() = default;
+  Connection(std::weak_ptr<observable_detail::HandlerTable> table, int id) : table_(std::move(table)), id_(id) {}
+  Connection(Connection &&other) noexcept { *this = std::move(other); }
+  Connection &operator=(Connection &&other) noexcept {
+    if (this != &other) {
+      disconnect();
+      table_ = std::move(other.table_);
+      id_ = std::exchange(other.id_, -1);
+    }
+    return *this;
+  }
+  Connection(const Connection &) = delete;
+  Connection &operator=(const Connection &) = delete;
+  ~Connection() { disconnect(); }
+
+  void disconnect() {
+    if (auto table = table_.lock()) table->erase(id_);
+    table_.reset();
+    id_ = -1;
+  }
+
+private:
+  std::weak_ptr<observable_detail::HandlerTable> table_;
+  int id_ = -1;
+};
+
+using Connections = std::vector<Connection>;
+
+// main thread only. handlers may disconnect (or destroy the Observable) while being invoked.
+template <typename... Args>
+class Observable {
+public:
+  using Handler = std::function<void(Args...)>;
+
+  Observable() = default;
+  Observable(const Observable &) = delete;
+  Observable &operator=(const Observable &) = delete;
+
+  [[nodiscard]] Connection connect(Handler handler) {
+    int id = table_->next_id++;
+    table_->handlers.emplace(id, std::make_shared<Handler>(std::move(handler)));
+    return Connection(table_, id);
+  }
+
+  void operator()(Args... args) const {
+    auto table = table_;
+    std::vector<int> ids;
+    ids.reserve(table->handlers.size());
+    for (const auto &[id, _] : table->handlers) ids.push_back(id);
+    for (int id : ids) {
+      auto it = table->handlers.find(id);
+      if (it == table->handlers.end()) continue;
+      auto handler = it->second;
+      (*handler)(args...);
+    }
+  }
+
+private:
+  struct Table : observable_detail::HandlerTable {
+    std::map<int, std::shared_ptr<Handler>> handlers;
+    int next_id = 0;
+    void erase(int id) override { handlers.erase(id); }
+  };
+  std::shared_ptr<Table> table_ = std::make_shared<Table>();
+};
diff --git a/openpilot/tools/cabana/dbc/dbcmanager.cc b/openpilot/tools/cabana/dbc/dbcmanager.cc
index 7a95a4f80..551619ea8 100644
--- a/openpilot/tools/cabana/dbc/dbcmanager.cc
+++ b/openpilot/tools/cabana/dbc/dbcmanager.cc
@@ -17,7 +17,7 @@ bool DBCManager::open(const SourceSet &sources, const std::string &dbc_file_name
     return false;
   }
 
-  if (callbacks_.file_changed) callbacks_.file_changed();
+  fileChanged();
   return true;
 }
 
@@ -32,7 +32,7 @@ bool DBCManager::open(const SourceSet &sources, const std::string &name, const s
     return false;
   }
 
-  if (callbacks_.file_changed) callbacks_.file_changed();
+  fileChanged();
   return true;
 }
 
@@ -40,26 +40,26 @@ void DBCManager::close(const SourceSet &sources) {
   for (auto s : sources) {
     dbc_files[s] = nullptr;
   }
-  if (callbacks_.file_changed) callbacks_.file_changed();
+  fileChanged();
 }
 
 void DBCManager::close(DBCFile *dbc_file) {
   for (auto &[_, f] : dbc_files) {
     if (f.get() == dbc_file) f = nullptr;
   }
-  if (callbacks_.file_changed) callbacks_.file_changed();
+  fileChanged();
 }
 
 void DBCManager::closeAll() {
   dbc_files.clear();
-  if (callbacks_.file_changed) callbacks_.file_changed();
+  fileChanged();
 }
 
 void DBCManager::addSignal(const MessageId &id, const cabana::Signal &sig) {
   if (auto m = msg(id)) {
     if (auto s = m->addSignal(sig)) {
-      if (callbacks_.signal_added) callbacks_.signal_added(id, s);
-      if (callbacks_.mask_updated) callbacks_.mask_updated();
+      signalAdded(id, s);
+      maskUpdated();
     }
   }
 }
@@ -67,8 +67,8 @@ void DBCManager::addSignal(const MessageId &id, const cabana::Signal &sig) {
 void DBCManager::updateSignal(const MessageId &id, const std::string &sig_name, const cabana::Signal &sig) {
   if (auto m = msg(id)) {
     if (auto s = m->updateSignal(sig_name, sig)) {
-      if (callbacks_.signal_updated) callbacks_.signal_updated(s);
-      if (callbacks_.mask_updated) callbacks_.mask_updated();
+      signalUpdated(s);
+      maskUpdated();
     }
   }
 }
@@ -76,9 +76,9 @@ void DBCManager::updateSignal(const MessageId &id, const std::string &sig_name,
 void DBCManager::removeSignal(const MessageId &id, const std::string &sig_name) {
   if (auto m = msg(id)) {
     if (auto s = m->sig(sig_name)) {
-      if (callbacks_.signal_removed) callbacks_.signal_removed(s);
+      signalRemoved(s);
       m->removeSignal(sig_name);
-      if (callbacks_.mask_updated) callbacks_.mask_updated();
+      maskUpdated();
     }
   }
 }
@@ -87,15 +87,15 @@ void DBCManager::updateMsg(const MessageId &id, const std::string &name, uint32_
   auto dbc_file = findDBCFile(id);
   assert(dbc_file);  // This should be impossible
   dbc_file->updateMsg(id, name, size, node, comment);
-  if (callbacks_.msg_updated) callbacks_.msg_updated(id);
+  msgUpdated(id);
 }
 
 void DBCManager::removeMsg(const MessageId &id) {
   auto dbc_file = findDBCFile(id);
   assert(dbc_file);  // This should be impossible
   dbc_file->removeMsg(id);
-  if (callbacks_.msg_removed) callbacks_.msg_removed(id);
-  if (callbacks_.mask_updated) callbacks_.mask_updated();
+  msgRemoved(id);
+  maskUpdated();
 }
 
 std::string DBCManager::newMsgName(const MessageId &id) {
diff --git a/openpilot/tools/cabana/dbc/dbcmanager.h b/openpilot/tools/cabana/dbc/dbcmanager.h
index 5a09fae03..6bc86a816 100644
--- a/openpilot/tools/cabana/dbc/dbcmanager.h
+++ b/openpilot/tools/cabana/dbc/dbcmanager.h
@@ -1,12 +1,12 @@
 #pragma once
 
-#include <functional>
 #include <memory>
 #include <map>
 #include <set>
 #include <string>
 #include <vector>
 
+#include "tools/cabana/core/observable.h"
 #include "tools/cabana/dbc/dbcfile.h"
 
 typedef std::set<int> SourceSet;
@@ -15,16 +15,6 @@ inline bool operator<(const std::shared_ptr<DBCFile> &l, const std::shared_ptr<D
 
 class DBCManager {
 public:
-  struct Callbacks {
-    std::function<void(MessageId, const cabana::Signal *)> signal_added;
-    std::function<void(const cabana::Signal *)> signal_removed;
-    std::function<void(const cabana::Signal *)> signal_updated;
-    std::function<void(MessageId)> msg_updated;
-    std::function<void(MessageId)> msg_removed;
-    std::function<void()> file_changed;
-    std::function<void()> mask_updated;
-  };
-
   DBCManager() = default;
   bool open(const SourceSet &sources, const std::string &dbc_file_name, std::string *error = nullptr);
   bool open(const SourceSet &sources, const std::string &name, const std::string &content, std::string *error = nullptr);
@@ -54,11 +44,17 @@ public:
   DBCFile *findDBCFile(const uint8_t source);
   inline DBCFile *findDBCFile(const MessageId &id) { return findDBCFile(id.source); }
   std::set<DBCFile *> allDBCFiles();
-  void setCallbacks(Callbacks callbacks) { callbacks_ = std::move(callbacks); }
+
+  Observable<MessageId, const cabana::Signal *> signalAdded;
+  Observable<const cabana::Signal *> signalRemoved;
+  Observable<const cabana::Signal *> signalUpdated;
+  Observable<MessageId> msgUpdated;
+  Observable<MessageId> msgRemoved;
+  Observable<> fileChanged;
+  Observable<> maskUpdated;
 
 private:
   std::map<int, std::shared_ptr<DBCFile>> dbc_files;
-  Callbacks callbacks_;
 };
 
 DBCManager *dbc();
diff --git a/openpilot/tools/cabana/dbc/dbcqt.cc b/openpilot/tools/cabana/dbc/dbcqt.cc
deleted file mode 100644
index 4354caf3f..000000000
--- a/openpilot/tools/cabana/dbc/dbcqt.cc
+++ /dev/null
@@ -1,18 +0,0 @@
-#include "tools/cabana/dbc/dbcqt.h"
-
-QtDBCNotifier::QtDBCNotifier(QObject *parent) : QObject(parent) {
-  dbc()->setCallbacks({
-    .signal_added = [this](MessageId id, const cabana::Signal *sig) { emit signalAdded(id, sig); },
-    .signal_removed = [this](const cabana::Signal *sig) { emit signalRemoved(sig); },
-    .signal_updated = [this](const cabana::Signal *sig) { emit signalUpdated(sig); },
-    .msg_updated = [this](MessageId id) { emit msgUpdated(id); },
-    .msg_removed = [this](MessageId id) { emit msgRemoved(id); },
-    .file_changed = [this]() { emit DBCFileChanged(); },
-    .mask_updated = [this]() { emit maskUpdated(); },
-  });
-}
-
-QtDBCNotifier *dbcNotifier() {
-  static QtDBCNotifier notifier;
-  return &notifier;
-}
diff --git a/openpilot/tools/cabana/dbc/dbcqt.h b/openpilot/tools/cabana/dbc/dbcqt.h
deleted file mode 100644
index b889f854e..000000000
--- a/openpilot/tools/cabana/dbc/dbcqt.h
+++ /dev/null
@@ -1,27 +0,0 @@
-#pragma once
-
-#include <QMetaType>
-#include <QObject>
-
-#include "tools/cabana/dbc/dbcmanager.h"
-
-Q_DECLARE_METATYPE(MessageId)
-Q_DECLARE_METATYPE(ValueDescription)
-
-class QtDBCNotifier : public QObject {
-  Q_OBJECT
-
-public:
-  explicit QtDBCNotifier(QObject *parent = nullptr);
-
-signals:
-  void signalAdded(MessageId id, const cabana::Signal *sig);
-  void signalRemoved(const cabana::Signal *sig);
-  void signalUpdated(const cabana::Signal *sig);
-  void msgUpdated(MessageId id);
-  void msgRemoved(MessageId id);
-  void DBCFileChanged();
-  void maskUpdated();
-};
-
-QtDBCNotifier *dbcNotifier();
diff --git a/openpilot/tools/cabana/detailwidget.cc b/openpilot/tools/cabana/detailwidget.cc
index 6b62f5495..d0cccb599 100644
--- a/openpilot/tools/cabana/detailwidget.cc
+++ b/openpilot/tools/cabana/detailwidget.cc
@@ -1,5 +1,4 @@
 #include "tools/cabana/detailwidget.h"
-#include "tools/cabana/dbc/dbcqt.h"
 
 #include <QFormLayout>
 #include <QMenu>
@@ -56,9 +55,9 @@ DetailWidget::DetailWidget(ChartsWidget *charts, QWidget *parent) : charts(chart
   QObject::connect(signal_view, &SignalView::showChart, charts, &ChartsWidget::showChart);
   QObject::connect(signal_view, &SignalView::highlight, binary_view, &BinaryView::highlight);
   QObject::connect(tab_widget, &QTabWidget::currentChanged, [this]() { updateState(); });
-  QObject::connect(can, &AbstractStream::msgsReceived, this, &DetailWidget::updateState);
-  QObject::connect(dbcNotifier(), &QtDBCNotifier::DBCFileChanged, this, &DetailWidget::refresh);
-  QObject::connect(undoNotifier(), &QtUndoNotifier::indexChanged, this, &DetailWidget::refresh);
+  connections_.push_back(can->msgsReceived.connect([this](const std::set<MessageId> *msgs, bool) { updateState(msgs); }));
+  connections_.push_back(dbc()->fileChanged.connect([this]() { refresh(); }));
+  connections_.push_back(UndoStack::instance()->indexChanged.connect([this]() { refresh(); }));
   QObject::connect(tabbar, &QTabBar::customContextMenuRequested, this, &DetailWidget::showTabBarContextMenu);
   QObject::connect(tabbar, &QTabBar::currentChanged, [this](int index) {
     if (index != -1) {
@@ -97,11 +96,11 @@ void DetailWidget::createToolBar() {
   layout()->addWidget(toolbar);
 
   connect(heatmap_live, &QAbstractButton::toggled, this, [this](bool on) { binary_view->setHeatmapLiveMode(on); });
-  connect(can, &AbstractStream::timeRangeChanged, this, [=](const std::optional<std::pair<double, double>> &range) {
+  connections_.push_back(can->timeRangeChanged.connect([=](const std::optional<std::pair<double, double>> &range) {
     auto text = range ? QString("%1 - %2").arg(range->first, 0, 'f', 3).arg(range->second, 0, 'f', 3) : "All";
     heatmap_all->setText(text);
     (range ? heatmap_all : heatmap_live)->setChecked(true);
-  });
+  }));
 }
 
 void DetailWidget::showTabBarContextMenu(const QPoint &pt) {
diff --git a/openpilot/tools/cabana/detailwidget.h b/openpilot/tools/cabana/detailwidget.h
index 0fe1535c7..48d320c40 100644
--- a/openpilot/tools/cabana/detailwidget.h
+++ b/openpilot/tools/cabana/detailwidget.h
@@ -57,6 +57,7 @@ private:
   SignalView *signal_view;
   ChartsWidget *charts;
   QSplitter *splitter;
+  Connections connections_;
 };
 
 class CenterWidget : public QWidget {
diff --git a/openpilot/tools/cabana/historylog.cc b/openpilot/tools/cabana/historylog.cc
index 3fd569c64..26c0f4168 100644
--- a/openpilot/tools/cabana/historylog.cc
+++ b/openpilot/tools/cabana/historylog.cc
@@ -1,5 +1,4 @@
 #include "tools/cabana/historylog.h"
-#include "tools/cabana/dbc/dbcqt.h"
 
 #include <functional>
 
@@ -10,6 +9,12 @@
 #include "tools/cabana/commands.h"
 #include "tools/cabana/utils/export.h"
 
+HistoryLogModel::HistoryLogModel(QObject *parent) : QAbstractTableModel(parent) {
+  connections_.push_back(can->seekedTo.connect([this](double) { reset(); }));
+  connections_.push_back(dbc()->fileChanged.connect([this]() { reset(); }));
+  connections_.push_back(UndoStack::instance()->indexChanged.connect([this]() { reset(); }));
+}
+
 QVariant HistoryLogModel::data(const QModelIndex &index, int role) const {
   const auto &m = messages[index.row()];
   const int col = index.column();
@@ -207,9 +212,6 @@ LogsWidget::LogsWidget(QWidget *parent) : QFrame(parent) {
   QObject::connect(comp_box, SIGNAL(activated(int)), this, SLOT(filterChanged()));
   QObject::connect(value_edit, &QLineEdit::textEdited, this, &LogsWidget::filterChanged);
   QObject::connect(export_btn, &QToolButton::clicked, this, &LogsWidget::exportToCSV);
-  QObject::connect(can, &AbstractStream::seekedTo, model, &HistoryLogModel::reset);
-  QObject::connect(dbcNotifier(), &QtDBCNotifier::DBCFileChanged, model, &HistoryLogModel::reset);
-  QObject::connect(undoNotifier(), &QtUndoNotifier::indexChanged, model, &HistoryLogModel::reset);
   QObject::connect(model, &HistoryLogModel::modelReset, this, &LogsWidget::modelReset);
   QObject::connect(model, &HistoryLogModel::rowsInserted, [this]() { export_btn->setEnabled(true); });
 }
diff --git a/openpilot/tools/cabana/historylog.h b/openpilot/tools/cabana/historylog.h
index 1d3200b20..405fe60fb 100644
--- a/openpilot/tools/cabana/historylog.h
+++ b/openpilot/tools/cabana/historylog.h
@@ -22,7 +22,7 @@ class HistoryLogModel : public QAbstractTableModel {
   Q_OBJECT
 
 public:
-  HistoryLogModel(QObject *parent) : QAbstractTableModel(parent) {}
+  HistoryLogModel(QObject *parent);
   void setMessage(const MessageId &message_id);
   void updateState(bool clear = false);
   void setFilter(int sig_idx, const QString &value, std::function<bool(double, double)> cmp);
@@ -54,6 +54,7 @@ public:
   std::deque<Message> messages;
   std::vector<cabana::Signal *> sigs;
   bool hex_mode = false;
+  Connections connections_;
 };
 
 class LogsWidget : public QFrame {
diff --git a/openpilot/tools/cabana/mainwin.cc b/openpilot/tools/cabana/mainwin.cc
index 953f573a0..bcfe1dde4 100644
--- a/openpilot/tools/cabana/mainwin.cc
+++ b/openpilot/tools/cabana/mainwin.cc
@@ -1,5 +1,4 @@
 #include "tools/cabana/mainwin.h"
-#include "tools/cabana/dbc/dbcqt.h"
 
 #include <algorithm>
 #include <filesystem>
@@ -41,15 +40,13 @@ MainWindow::MainWindow(AbstractStream *stream, const QString &dbc_file) : QMainW
   restoreGeometry(utils::qbytes(settings.geometry));
   restoreState(utils::qbytes(settings.window_state));
 
-  // install handlers
+  // download handlers are called from download threads
   static auto static_main_win = this;
-  qRegisterMetaType<uint64_t>("uint64_t");
-  qRegisterMetaType<SourceSet>("SourceSet");
   installDownloadProgressHandler([](uint64_t cur, uint64_t total, bool success) {
-    emit static_main_win->updateProgressBar(cur, total, success);
+    utils::runOnMainThread([=]() { static_main_win->updateDownloadProgress(cur, total, success); });
   });
   installMessageHandler([](ReplyMsgType type, const std::string msg) {
-    emit static_main_win->showMessage(QString::fromStdString(msg), 2000);
+    utils::runOnMainThread([=]() { static_main_win->statusBar()->showMessage(QString::fromStdString(msg), 2000); });
   });
 
   setStyleSheet(QString(R"(QMainWindow::separator {
@@ -57,11 +54,14 @@ MainWindow::MainWindow(AbstractStream *stream, const QString &dbc_file) : QMainW
     height: %1px; /* when horizontal */
   })").arg(style()->pixelMetric(QStyle::PM_SplitterWidth)));
 
-  QObject::connect(this, &MainWindow::showMessage, statusBar(), &QStatusBar::showMessage);
-  QObject::connect(this, &MainWindow::updateProgressBar, this, &MainWindow::updateDownloadProgress);
-  QObject::connect(dbcNotifier(), &QtDBCNotifier::DBCFileChanged, this, &MainWindow::DBCFileChanged);
-  QObject::connect(undoNotifier(), &QtUndoNotifier::cleanChanged, this, &MainWindow::undoStackCleanChanged);
-  QObject::connect(&settings, &Settings::changed, this, &MainWindow::updateStatus);
+  connections_.push_back(dbc()->fileChanged.connect([this]() { DBCFileChanged(); }));
+  connections_.push_back(UndoStack::instance()->cleanChanged.connect([this](bool clean) { undoStackCleanChanged(clean); }));
+  connections_.push_back(settings.changed.connect([this]() { updateStatus(); }));
+
+  // temporary pump for the non-Qt main thread queue until imgui owns the loop
+  auto *queue_timer = new QTimer(this);
+  QObject::connect(queue_timer, &QTimer::timeout, utils::drainMainThreadQueue);
+  queue_timer->start(10);
 
   QTimer::singleShot(0, this, [=]() { stream ? openStream(stream, dbc_file) : selectAndOpenStream(); });
   show();
@@ -136,7 +136,7 @@ void MainWindow::createActions() {
   undo_act->setShortcuts(QKeySequence::Undo);
   redo_act = edit_menu->addAction(tr("&Redo"), []() { UndoStack::instance()->redo(); });
   redo_act->setShortcuts(QKeySequence::Redo);
-  QObject::connect(undoNotifier(), &QtUndoNotifier::indexChanged, this, &MainWindow::updateUndoRedoActions);
+  connections_.push_back(UndoStack::instance()->indexChanged.connect([this]() { updateUndoRedoActions(); }));
   updateUndoRedoActions();
 
   // View Menu
@@ -259,14 +259,14 @@ void MainWindow::selectAndOpenStream() {
   if (dlg.exec()) {
     openStream(dlg.stream(), dlg.dbcFile());
   } else if (!can) {
-    openStream(new DummyStream(this));
+    openStream(new DummyStream());
   }
 }
 
 void MainWindow::closeStream() {
-  openStream(new DummyStream(this));
+  openStream(new DummyStream());
   if (dbc()->nonEmptyDBCCount() > 0) {
-    emit dbcNotifier()->DBCFileChanged();
+    dbc()->fileChanged();
   }
   statusBar()->showMessage(tr("stream closed"));
 }
@@ -336,13 +336,19 @@ void MainWindow::loadFromClipboard(SourceSet s, bool close_all) {
   }
 }
 
+// stream threads read the global `can` until its destructor joins them
+MainWindow::~MainWindow() {
+  delete can;
+  can = nullptr;
+}
+
 void MainWindow::openStream(AbstractStream *stream, const QString &dbc_file) {
-  if (can) {
-    QObject::connect(can, &QObject::destroyed, this, [=]() { startStream(stream, dbc_file); });
-    can->deleteLater();
-  } else {
-    startStream(stream, dbc_file);
-  }
+  stream_connections_.clear();
+  if (wait_dlg_) wait_dlg_->deleteLater();
+  wait_dlg_ = nullptr;
+  delete can;
+  can = nullptr;
+  startStream(stream, dbc_file);
 }
 
 void MainWindow::startStream(AbstractStream *stream, QString dbc_file) {
@@ -350,8 +356,7 @@ void MainWindow::startStream(AbstractStream *stream, QString dbc_file) {
   delete messages_widget;
   delete video_splitter;
 
-  can = stream;
-  can->setParent(this);  // take ownership
+  can = stream;  // take ownership
   can->start();
 
   loadFile(dbc_file);
@@ -373,18 +378,19 @@ void MainWindow::startStream(AbstractStream *stream, QString dbc_file) {
     newFile();
   }
 
-  QObject::connect(can, &AbstractStream::eventsMerged, this, &MainWindow::eventsMerged);
+  stream_connections_.push_back(can->eventsMerged.connect([this](const MessageEventsMap &) { eventsMerged(); }));
 
   if (has_stream) {
-    auto wait_dlg = new QProgressDialog(
+    wait_dlg_ = new QProgressDialog(
         can->liveStreaming() ? tr("Waiting for the live stream to start...") : tr("Loading segment data..."),
         tr("&Abort"), 0, 100, this);
-    wait_dlg->setWindowModality(Qt::WindowModal);
-    wait_dlg->setFixedSize(400, wait_dlg->sizeHint().height());
-    QObject::connect(wait_dlg, &QProgressDialog::canceled, this, &MainWindow::close);
-    QObject::connect(can, &AbstractStream::eventsMerged, wait_dlg, &QProgressDialog::deleteLater);
-    QObject::connect(this, &MainWindow::updateProgressBar, wait_dlg, [=](uint64_t cur, uint64_t total, bool success) {
-      wait_dlg->setValue((int)((cur / (double)total) * 100));
+    wait_dlg_->setWindowModality(Qt::WindowModal);
+    wait_dlg_->setFixedSize(400, wait_dlg_->sizeHint().height());
+    QObject::connect(wait_dlg_, &QProgressDialog::canceled, this, &MainWindow::close);
+    wait_dlg_connection_ = can->eventsMerged.connect([this](const MessageEventsMap &) {
+      wait_dlg_->deleteLater();
+      wait_dlg_ = nullptr;
+      wait_dlg_connection_.disconnect();
     });
   }
 }
@@ -544,6 +550,7 @@ void MainWindow::remindSaveChanges() {
 }
 
 void MainWindow::updateDownloadProgress(uint64_t cur, uint64_t total, bool success) {
+  if (wait_dlg_) wait_dlg_->setValue((int)((cur / (double)total) * 100));
   if (success && cur < total) {
     progress_bar->setValue((cur / (double)total) * 100);
     progress_bar->setFormat(tr("Downloading %p% (%1)").arg(formattedDataSize(total).c_str()));
diff --git a/openpilot/tools/cabana/mainwin.h b/openpilot/tools/cabana/mainwin.h
index 279ea3b96..a57232ceb 100644
--- a/openpilot/tools/cabana/mainwin.h
+++ b/openpilot/tools/cabana/mainwin.h
@@ -19,11 +19,14 @@
 #include "tools/cabana/videowidget.h"
 #include "tools/cabana/tools/findsimilarbits.h"
 
+class QProgressDialog;
+
 class MainWindow : public QMainWindow {
   Q_OBJECT
 
 public:
   MainWindow(AbstractStream *stream, const QString &dbc_file);
+  ~MainWindow();
   void toggleChartsDocking();
   void showStatusMessage(const QString &msg, int timeout = 0) { statusBar()->showMessage(msg, timeout); }
   void loadFile(const QString &fn, SourceSet s = SOURCE_ALL);
@@ -42,10 +45,6 @@ public slots:
   void saveAs();
   void saveToClipboard();
 
-signals:
-  void showMessage(const QString &msg, int timeout);
-  void updateProgressBar(uint64_t cur, uint64_t total, bool success);
-
 protected:
   void startStream(AbstractStream *stream, QString dbc_file);
   bool eventFilter(QObject *obj, QEvent *event) override;
@@ -104,6 +103,10 @@ protected:
   QAction *redo_act = nullptr;
   QString car_fingerprint;
   std::vector<uint8_t> default_state;
+  Connections connections_;
+  Connections stream_connections_;
+  Connection wait_dlg_connection_;
+  QProgressDialog *wait_dlg_ = nullptr;
 };
 
 class HelpOverlay : public QWidget {
diff --git a/openpilot/tools/cabana/messageswidget.cc b/openpilot/tools/cabana/messageswidget.cc
index b07e3ea0b..eea07b871 100644
--- a/openpilot/tools/cabana/messageswidget.cc
+++ b/openpilot/tools/cabana/messageswidget.cc
@@ -1,5 +1,4 @@
 #include "tools/cabana/messageswidget.h"
-#include "tools/cabana/dbc/dbcqt.h"
 
 #include <limits>
 #include <utility>
@@ -43,9 +42,6 @@ MessagesWidget::MessagesWidget(QWidget *parent) : menu(new QMenu(this)), QWidget
   QObject::connect(menu, &QMenu::aboutToShow, this, &MessagesWidget::menuAboutToShow);
   QObject::connect(header, &MessageViewHeader::customContextMenuRequested, this, &MessagesWidget::headerContextMenuEvent);
   QObject::connect(view->horizontalScrollBar(), &QScrollBar::valueChanged, header, &MessageViewHeader::updateHeaderPositions);
-  QObject::connect(can, &AbstractStream::msgsReceived, model, &MessageListModel::msgsReceived);
-  QObject::connect(dbcNotifier(), &QtDBCNotifier::DBCFileChanged, model, &MessageListModel::dbcModified);
-  QObject::connect(undoNotifier(), &QtUndoNotifier::indexChanged, model, &MessageListModel::dbcModified);
   QObject::connect(model, &MessageListModel::modelReset, [this]() {
     if (current_msg_id) {
       selectMessage(*current_msg_id);
@@ -96,7 +92,7 @@ QWidget *MessagesWidget::createToolBar() {
 
   QObject::connect(suppress_add, &QPushButton::clicked, this, &MessagesWidget::suppressHighlighted);
   QObject::connect(suppress_clear, &QPushButton::clicked, this, &MessagesWidget::suppressHighlighted);
-  QObject::connect(suppress_defined_signals, &QCheckBox::stateChanged, can, &AbstractStream::suppressDefinedSignals);
+  QObject::connect(suppress_defined_signals, &QCheckBox::stateChanged, this, [](int state) { can->suppressDefinedSignals(state); });
 
   suppressHighlighted();
   return toolbar;
@@ -161,6 +157,12 @@ void MessagesWidget::setMultiLineBytes(bool multi) {
 
 // MessageListModel
 
+MessageListModel::MessageListModel(QObject *parent) : QAbstractTableModel(parent) {
+  connections_.push_back(can->msgsReceived.connect([this](const std::set<MessageId> *msgs, bool has_new_ids) { msgsReceived(msgs, has_new_ids); }));
+  connections_.push_back(dbc()->fileChanged.connect([this]() { dbcModified(); }));
+  connections_.push_back(UndoStack::instance()->indexChanged.connect([this]() { dbcModified(); }));
+}
+
 QVariant MessageListModel::headerData(int section, Qt::Orientation orientation, int role) const {
   if (orientation == Qt::Horizontal && role == Qt::DisplayRole) {
     switch (section) {
diff --git a/openpilot/tools/cabana/messageswidget.h b/openpilot/tools/cabana/messageswidget.h
index 0a9cd256d..28cab2546 100644
--- a/openpilot/tools/cabana/messageswidget.h
+++ b/openpilot/tools/cabana/messageswidget.h
@@ -31,7 +31,7 @@ public:
     DATA,
   };
 
-  MessageListModel(QObject *parent) : QAbstractTableModel(parent) {}
+  MessageListModel(QObject *parent);
   QVariant headerData(int section, Qt::Orientation orientation, int role = Qt::DisplayRole) const override;
   int columnCount(const QModelIndex &parent = QModelIndex()) const override { return Column::DATA + 1; }
   QVariant data(const QModelIndex &index, int role = Qt::DisplayRole) const;
@@ -63,6 +63,7 @@ private:
   int sort_column = 0;
   Qt::SortOrder sort_order = Qt::AscendingOrder;
   int sort_threshold_ = 0;
+  Connections connections_;
 };
 
 class MessageView : public QTreeView {
diff --git a/openpilot/tools/cabana/settings.cc b/openpilot/tools/cabana/settings.cc
index 0c8136b84..cbeded0d1 100644
--- a/openpilot/tools/cabana/settings.cc
+++ b/openpilot/tools/cabana/settings.cc
@@ -613,6 +613,6 @@ void SettingsDlg::save() {
   settings.log_livestream = log_livestream->isChecked();
   settings.log_path = log_path->text().toStdString();
   settings.drag_direction = (Settings::DragDirection)drag_direction->currentIndex();
-  emit settings.changed();
+  settings.changed();
   QDialog::accept();
 }
diff --git a/openpilot/tools/cabana/settings.h b/openpilot/tools/cabana/settings.h
index 7357ecf4f..2c21fab41 100644
--- a/openpilot/tools/cabana/settings.h
+++ b/openpilot/tools/cabana/settings.h
@@ -9,11 +9,10 @@
 #include <QLineEdit>
 #include <QSpinBox>
 
+#include "tools/cabana/core/observable.h"
 #include "tools/cabana/core/settings.h"
 
-class Settings : public QObject, public CabanaSettingsState {
-  Q_OBJECT
-
+class Settings : public CabanaSettingsState {
 public:
   Settings();
   void save();
@@ -24,8 +23,7 @@ public:
   std::vector<uint8_t> window_state;
   std::vector<uint8_t> message_header_state;
 
-signals:
-  void changed();
+  Observable<> changed;
 };
 
 class SettingsDlg : public QDialog {
diff --git a/openpilot/tools/cabana/signalview.cc b/openpilot/tools/cabana/signalview.cc
index ebb537414..250d05e3b 100644
--- a/openpilot/tools/cabana/signalview.cc
+++ b/openpilot/tools/cabana/signalview.cc
@@ -1,5 +1,4 @@
 #include "tools/cabana/signalview.h"
-#include "tools/cabana/dbc/dbcqt.h"
 
 #include <algorithm>
 #include <future>
@@ -26,12 +25,12 @@ static QString signalTypeToString(cabana::Signal::Type type) {
 }
 
 SignalModel::SignalModel(QObject *parent) : root(new Item), QAbstractItemModel(parent) {
-  QObject::connect(dbcNotifier(), &QtDBCNotifier::DBCFileChanged, this, &SignalModel::refresh);
-  QObject::connect(dbcNotifier(), &QtDBCNotifier::msgUpdated, this, &SignalModel::handleMsgChanged);
-  QObject::connect(dbcNotifier(), &QtDBCNotifier::msgRemoved, this, &SignalModel::handleMsgChanged);
-  QObject::connect(dbcNotifier(), &QtDBCNotifier::signalAdded, this, &SignalModel::handleSignalAdded);
-  QObject::connect(dbcNotifier(), &QtDBCNotifier::signalUpdated, this, &SignalModel::handleSignalUpdated);
-  QObject::connect(dbcNotifier(), &QtDBCNotifier::signalRemoved, this, &SignalModel::handleSignalRemoved);
+  connections_.push_back(dbc()->fileChanged.connect([this]() { refresh(); }));
+  connections_.push_back(dbc()->msgUpdated.connect([this](MessageId id) { handleMsgChanged(id); }));
+  connections_.push_back(dbc()->msgRemoved.connect([this](MessageId id) { handleMsgChanged(id); }));
+  connections_.push_back(dbc()->signalAdded.connect([this](MessageId id, const cabana::Signal *sig) { handleSignalAdded(id, sig); }));
+  connections_.push_back(dbc()->signalUpdated.connect([this](const cabana::Signal *sig) { handleSignalUpdated(sig); }));
+  connections_.push_back(dbc()->signalRemoved.connect([this](const cabana::Signal *sig) { handleSignalRemoved(sig); }));
 }
 
 void SignalModel::insertItem(SignalModel::Item *root_item, int pos, const cabana::Signal *sig) {
@@ -472,11 +471,11 @@ SignalView::SignalView(ChartsWidget *charts, QWidget *parent) : charts(charts),
   QObject::connect(tree, &QTreeView::entered, [this](const QModelIndex &index) { emit highlight(model->getItem(index)->sig); });
   QObject::connect(model, &QAbstractItemModel::modelReset, this, &SignalView::rowsChanged);
   QObject::connect(model, &QAbstractItemModel::rowsRemoved, this, &SignalView::rowsChanged);
-  QObject::connect(dbcNotifier(), &QtDBCNotifier::signalAdded, this, &SignalView::handleSignalAdded);
-  QObject::connect(dbcNotifier(), &QtDBCNotifier::signalUpdated, this, &SignalView::handleSignalUpdated);
+  connections_.push_back(dbc()->signalAdded.connect([this](MessageId id, const cabana::Signal *sig) { handleSignalAdded(id, sig); }));
+  connections_.push_back(dbc()->signalUpdated.connect([this](const cabana::Signal *sig) { handleSignalUpdated(sig); }));
   QObject::connect(tree->verticalScrollBar(), &QScrollBar::valueChanged, [this]() { updateState(); });
   QObject::connect(tree->verticalScrollBar(), &QScrollBar::rangeChanged, [this]() { updateState(); });
-  QObject::connect(can, &AbstractStream::msgsReceived, this, &SignalView::updateState);
+  connections_.push_back(can->msgsReceived.connect([this](const std::set<MessageId> *msgs, bool) { updateState(msgs); }));
   QObject::connect(tree->header(), &QHeaderView::sectionResized, [this](int logicalIndex, int oldSize, int newSize) {
     if (logicalIndex == 1) {
       value_column_width = newSize;
diff --git a/openpilot/tools/cabana/signalview.h b/openpilot/tools/cabana/signalview.h
index bccfbab0c..e2a99ebbd 100644
--- a/openpilot/tools/cabana/signalview.h
+++ b/openpilot/tools/cabana/signalview.h
@@ -62,6 +62,7 @@ private:
   MessageId msg_id;
   QString filter_str;
   std::unique_ptr<Item> root;
+  Connections connections_;
   friend class SignalView;
   friend class SignalItemDelegate;
 };
@@ -150,4 +151,5 @@ private:
   ChartsWidget *charts;
   QLabel *signal_count_lb;
   SignalItemDelegate *delegate;
+  Connections connections_;
 };
diff --git a/openpilot/tools/cabana/streams/abstractstream.cc b/openpilot/tools/cabana/streams/abstractstream.cc
index c58a98084..cc8b2f96a 100644
--- a/openpilot/tools/cabana/streams/abstractstream.cc
+++ b/openpilot/tools/cabana/streams/abstractstream.cc
@@ -1,10 +1,8 @@
 #include "tools/cabana/streams/abstractstream.h"
-#include "tools/cabana/dbc/dbcqt.h"
 
 #include <limits>
 #include <utility>
 
-#include <QApplication>
 #include "common/timing.h"
 #include "tools/cabana/settings.h"
 
@@ -12,15 +10,41 @@ static const int EVENT_NEXT_BUFFER_SIZE = 6 * 1024 * 1024;  // 6MB
 
 AbstractStream *can = nullptr;
 
-AbstractStream::AbstractStream(QObject *parent) : QObject(parent) {
-  assert(parent != nullptr);
+AbstractStream::AbstractStream() {
   event_buffer_ = std::make_unique<MonotonicBuffer>(EVENT_NEXT_BUFFER_SIZE);
 
-  QObject::connect(this, &AbstractStream::privateUpdateLastMsgsSignal, this, &AbstractStream::updateLastMessages, Qt::QueuedConnection);
-  QObject::connect(this, &AbstractStream::seekedTo, this, &AbstractStream::updateLastMsgsTo);
-  QObject::connect(this, &AbstractStream::seeking, this, [this](double sec) { current_sec_ = sec; });
-  QObject::connect(dbcNotifier(), &QtDBCNotifier::DBCFileChanged, this, &AbstractStream::updateMasks);
-  QObject::connect(dbcNotifier(), &QtDBCNotifier::maskUpdated, this, &AbstractStream::updateMasks);
+  // connected first so the stream state is updated before any widget handlers run
+  connections_.push_back(seekedTo.connect([this](double sec) { updateLastMsgsTo(sec); }));
+  connections_.push_back(seeking.connect([this](double sec) { current_sec_ = sec; }));
+  connections_.push_back(dbc()->fileChanged.connect([this]() { updateMasks(); }));
+  connections_.push_back(dbc()->maskUpdated.connect([this]() { updateMasks(); }));
+}
+
+void AbstractStream::postToMainThread(std::function<void()> fn) {
+  utils::runOnMainThread([alive = std::weak_ptr<bool>(alive_), fn = std::move(fn)]() {
+    if (!alive.expired()) fn();
+  });
+}
+
+void AbstractStream::postToMainThreadAndWait(std::function<void()> fn) {
+  assert(!utils::isMainThread());
+  std::unique_lock lock(mutex_);
+  if (exiting_) return;
+  auto done = std::make_shared<bool>(false);
+  postToMainThread([this, alive = std::weak_ptr<bool>(alive_), done, fn = std::move(fn)]() {
+    fn();
+    if (alive.expired()) return;  // fn deleted the stream, the waiter was released by cancelWaits()
+    std::lock_guard lk(mutex_);
+    *done = true;
+    wait_cv_.notify_all();
+  });
+  wait_cv_.wait(lock, [&]() { return *done || exiting_; });
+}
+
+void AbstractStream::cancelWaits() {
+  std::lock_guard lk(mutex_);
+  exiting_ = true;
+  wait_cv_.notify_all();
 }
 
 void AbstractStream::updateMasks() {
@@ -97,9 +121,8 @@ void AbstractStream::updateLastMessages() {
 
   if (sources.size() != prev_src_size) {
     updateMasks();
-    emit sourcesUpdated(sources);
   }
-  emit msgsReceived(&msgs, prev_msg_size != last_msgs.size());
+  msgsReceived(&msgs, prev_msg_size != last_msgs.size());
 }
 
 void AbstractStream::setTimeRange(const std::optional<std::pair<double, double>> &range) {
@@ -107,7 +130,7 @@ void AbstractStream::setTimeRange(const std::optional<std::pair<double, double>>
   if (time_range_ && (current_sec_ < time_range_->first || current_sec_ >= time_range_->second)) {
     seekTo(time_range_->first);
   }
-  emit timeRangeChanged(time_range_);
+  timeRangeChanged(time_range_);
 }
 
 void AbstractStream::updateEvent(const MessageId &id, double sec, const uint8_t *data, uint8_t size) {
@@ -175,16 +198,16 @@ void AbstractStream::updateLastMsgsTo(double sec) {
                     std::any_of(messages_.cbegin(), messages_.cend(),
                                 [this](const auto &m) { return !last_msgs.count(m.first); });
   last_msgs = messages_;
-  emit msgsReceived(nullptr, id_changed);
+  msgsReceived(nullptr, id_changed);
 
   std::lock_guard lk(mutex_);
   seek_finished_ = true;
-  seek_finished_cv_.notify_one();
+  wait_cv_.notify_all();
 }
 
 void AbstractStream::waitForSeekFinshed() {
   std::unique_lock lock(mutex_);
-  seek_finished_cv_.wait(lock, [this]() { return seek_finished_; });
+  wait_cv_.wait(lock, [this]() { return seek_finished_ || exiting_; });
   seek_finished_ = false;
 }
 
@@ -218,16 +241,16 @@ void AbstractStream::mergeEvents(const std::vector<const CanEvent *> &events) {
     }
     auto pos = std::upper_bound(all_events_.cbegin(), all_events_.cend(), events.front()->mono_time, CompareCanEvent());
     all_events_.insert(pos, events.cbegin(), events.cend());
-    emit eventsMerged(msg_events);
+    eventsMerged(msg_events);
   }
 }
 
 std::pair<CanEventIter, CanEventIter> AbstractStream::eventsInRange(const MessageId &id, std::optional<std::pair<double, double>> time_range) const {
-  const auto &events = can->events(id);
+  const auto &events = this->events(id);
   if (!time_range) return {events.begin(), events.end()};
 
-  auto first = std::lower_bound(events.begin(), events.end(), can->toMonoTime(time_range->first), CompareCanEvent());
-  auto last = std::upper_bound(first, events.end(), can->toMonoTime(time_range->second), CompareCanEvent());
+  auto first = std::lower_bound(events.begin(), events.end(), toMonoTime(time_range->first), CompareCanEvent());
+  auto last = std::upper_bound(first, events.end(), toMonoTime(time_range->second), CompareCanEvent());
   return {first, last};
 }
 
diff --git a/openpilot/tools/cabana/streams/abstractstream.h b/openpilot/tools/cabana/streams/abstractstream.h
index 82cb89993..17d212041 100644
--- a/openpilot/tools/cabana/streams/abstractstream.h
+++ b/openpilot/tools/cabana/streams/abstractstream.h
@@ -4,6 +4,7 @@
 #include <array>
 #include <condition_variable>
 #include <chrono>
+#include <functional>
 #include <memory>
 #include <mutex>
 #include <optional>
@@ -14,16 +15,15 @@
 
 #include "openpilot/cereal/messaging/messaging.h"
 #include "tools/cabana/core/can_data.h"
+#include "tools/cabana/core/observable.h"
 #include "tools/cabana/dbc/dbcmanager.h"
 #include "tools/cabana/utils/util.h"
 #include "tools/replay/util.h"
 
-class AbstractStream : public QObject {
-  Q_OBJECT
-
+class AbstractStream {
 public:
-  AbstractStream(QObject *parent);
-  virtual ~AbstractStream() {}
+  AbstractStream();
+  virtual ~AbstractStream() = default;
   virtual void start() = 0;
   virtual bool liveStreaming() const { return true; }
   virtual void seekTo(double ts) {}
@@ -56,21 +56,22 @@ public:
   void clearSuppressed();
   void suppressDefinedSignals(bool suppress);
 
-signals:
-  void paused();
-  void resume();
-  void seeking(double sec);
-  void seekedTo(double sec);
-  void timeRangeChanged(const std::optional<std::pair<double, double>> &range);
-  void eventsMerged(const MessageEventsMap &events_map);
-  void msgsReceived(const std::set<MessageId> *new_msgs, bool has_new_ids);
-  void sourcesUpdated(const SourceSet &s);
-  void privateUpdateLastMsgsSignal();
+  // invoked on the main thread
+  Observable<> paused;
+  Observable<> resume;
+  Observable<double> seeking;
+  Observable<double> seekedTo;
+  Observable<const std::optional<std::pair<double, double>> &> timeRangeChanged;
+  Observable<const MessageEventsMap &> eventsMerged;
+  Observable<const std::set<MessageId> *, bool> msgsReceived;
 
-public:
   SourceSet sources;
 
 protected:
+  void postToMainThread(std::function<void()> fn);  // dropped if the stream is destroyed first
+  void postToMainThreadAndWait(std::function<void()> fn);
+  void cancelWaits();  // call before joining threads, the main thread isn't pumping events during destruction
+  void requestUpdateLastMessages() { postToMainThread([this]() { updateLastMessages(); }); }
   void mergeEvents(const std::vector<const CanEvent *> &events);
   const CanEvent *newEvent(uint64_t mono_time, const cereal::CanData::Reader &c);
   void updateEvent(const MessageId &id, double sec, const uint8_t *data, uint8_t size);
@@ -87,11 +88,14 @@ private:
   MessageEventsMap events_;
   std::unordered_map<MessageId, CanData> last_msgs;
   std::unique_ptr<MonotonicBuffer> event_buffer_;
+  std::shared_ptr<bool> alive_ = std::make_shared<bool>(true);
+  Connections connections_;
 
   // Members accessed in multiple threads. (mutex protected)
   std::mutex mutex_;
-  std::condition_variable seek_finished_cv_;
+  std::condition_variable wait_cv_;
   bool seek_finished_ = false;
+  bool exiting_ = false;
   std::set<MessageId> new_msgs_;
   std::unordered_map<MessageId, CanData> messages_;
   std::unordered_map<MessageId, std::vector<uint8_t>> masks_;
@@ -108,9 +112,7 @@ signals:
 };
 
 class DummyStream : public AbstractStream {
-  Q_OBJECT
 public:
-  DummyStream(QObject *parent) : AbstractStream(parent) {}
   std::string routeName() const override { return "No Stream"; }
   void start() override {}
 };
diff --git a/openpilot/tools/cabana/streams/devicestream.cc b/openpilot/tools/cabana/streams/devicestream.cc
index 3bd51c079..3b4b54707 100644
--- a/openpilot/tools/cabana/streams/devicestream.cc
+++ b/openpilot/tools/cabana/streams/devicestream.cc
@@ -23,7 +23,7 @@
 
 // DeviceStream
 
-DeviceStream::DeviceStream(QObject *parent, QString address) : zmq_address(address), LiveStream(parent) {
+DeviceStream::DeviceStream(QString address) : zmq_address(address) {
 }
 
 DeviceStream::~DeviceStream() {
@@ -61,8 +61,8 @@ void DeviceStream::start() {
     // fails, the child writes errno and the parent aborts stream start.
     int err_pipe[2] = {-1, -1};
     if (::pipe(err_pipe) != 0) {
-      QMessageBox::warning(nullptr, tr("Error"),
-                           tr("Failed to start bridge: %1").arg(QString::fromLocal8Bit(strerror(errno))));
+      QMessageBox::warning(nullptr, "Error",
+                           QString("Failed to start bridge: %1").arg(QString::fromLocal8Bit(strerror(errno))));
       return;
     }
 
@@ -79,8 +79,8 @@ void DeviceStream::start() {
     ::close(err_pipe[1]);
     if (pid < 0) {
       ::close(err_pipe[0]);
-      QMessageBox::warning(nullptr, tr("Error"),
-                           tr("Failed to start bridge: %1").arg(QString::fromLocal8Bit(strerror(errno))));
+      QMessageBox::warning(nullptr, "Error",
+                           QString("Failed to start bridge: %1").arg(QString::fromLocal8Bit(strerror(errno))));
       return;
     }
 
@@ -91,8 +91,8 @@ void DeviceStream::start() {
       // Child failed to exec; reap and surface the error.
       int status = 0;
       ::waitpid(pid, &status, 0);
-      QMessageBox::warning(nullptr, tr("Error"),
-                           tr("Failed to start bridge: %1").arg(QString::fromLocal8Bit(strerror(exec_errno))));
+      QMessageBox::warning(nullptr, "Error",
+                           QString("Failed to start bridge: %1").arg(QString::fromLocal8Bit(strerror(exec_errno))));
       return;
     }
 
@@ -144,5 +144,5 @@ OpenDeviceWidget::OpenDeviceWidget(QWidget *parent) : AbstractOpenStreamWidget(p
 AbstractStream *OpenDeviceWidget::open() {
   QString ip = ip_address->text().isEmpty() ? "127.0.0.1" : ip_address->text();
   bool msgq = group->checkedId() == 0;
-  return new DeviceStream(qApp, msgq ? "" : ip);
+  return new DeviceStream(msgq ? "" : ip);
 }
diff --git a/openpilot/tools/cabana/streams/devicestream.h b/openpilot/tools/cabana/streams/devicestream.h
index 0e6951c92..1aa8ce9bf 100644
--- a/openpilot/tools/cabana/streams/devicestream.h
+++ b/openpilot/tools/cabana/streams/devicestream.h
@@ -5,9 +5,8 @@
 #include <sys/types.h>
 
 class DeviceStream : public LiveStream {
-  Q_OBJECT
 public:
-  DeviceStream(QObject *parent, QString address = {});
+  DeviceStream(QString address = {});
   ~DeviceStream();
   inline std::string routeName() const override {
     return "Live Streaming From " + (zmq_address.isEmpty() ? std::string("127.0.0.1") : zmq_address.toStdString());
diff --git a/openpilot/tools/cabana/streams/livestream.cc b/openpilot/tools/cabana/streams/livestream.cc
index 019a67e8f..05a6d44c4 100644
--- a/openpilot/tools/cabana/streams/livestream.cc
+++ b/openpilot/tools/cabana/streams/livestream.cc
@@ -38,7 +38,7 @@ struct LiveStream::Logger {
   uint64_t start_ts;
 };
 
-LiveStream::LiveStream(QObject *parent) : AbstractStream(parent) {
+LiveStream::LiveStream() {
   if (settings.log_livestream) {
     logger = std::make_unique<Logger>();
   }
@@ -65,9 +65,9 @@ void LiveStream::stop() {
 void LiveStream::updateThread() {
   while (!exit_) {
     std::this_thread::sleep_for(std::chrono::milliseconds(1000 / fps_));
-    // coalesce: skip the emit if the main thread hasn't processed the previous one yet.
+    // coalesce: skip the request if the main thread hasn't processed the previous one yet.
     if (!update_pending_.exchange(true)) {
-      emit privateUpdateLastMsgsSignal();
+      requestUpdateLastMessages();
     }
   }
 }
@@ -89,7 +89,7 @@ void LiveStream::handleEvent(kj::ArrayPtr<capnp::word> data) {
   }
 }
 
-// called on the main thread by the queued privateUpdateLastMsgsSignal connection
+// called on the main thread via requestUpdateLastMessages()
 void LiveStream::updateLastMessages() {
   update_pending_ = false;
   fps_ = settings.fps;
@@ -142,10 +142,10 @@ void LiveStream::seekTo(double sec) {
   first_update_ts = nanos_since_boot();
   current_event_ts = first_event_ts = std::min<uint64_t>(sec * 1e9 + begin_event_ts, lastest_event_ts);
   post_last_event = (first_event_ts == lastest_event_ts);
-  emit seekedTo((current_event_ts - begin_event_ts) / 1e9);
+  seekedTo((current_event_ts - begin_event_ts) / 1e9);
 }
 
 void LiveStream::pause(bool pause) {
   paused_ = pause;
-  emit(pause ? paused() : resume());
+  pause ? paused() : resume();
 }
diff --git a/openpilot/tools/cabana/streams/livestream.h b/openpilot/tools/cabana/streams/livestream.h
index 5d65b1743..22587335b 100644
--- a/openpilot/tools/cabana/streams/livestream.h
+++ b/openpilot/tools/cabana/streams/livestream.h
@@ -9,10 +9,8 @@
 #include "tools/cabana/streams/abstractstream.h"
 
 class LiveStream : public AbstractStream {
-  Q_OBJECT
-
 public:
-  LiveStream(QObject *parent);
+  LiveStream();
   virtual ~LiveStream();
   void start() override;
   void stop();
diff --git a/openpilot/tools/cabana/streams/pandastream.cc b/openpilot/tools/cabana/streams/pandastream.cc
index 7ccb18a75..06c8e6e19 100644
--- a/openpilot/tools/cabana/streams/pandastream.cc
+++ b/openpilot/tools/cabana/streams/pandastream.cc
@@ -10,7 +10,7 @@
 #include <QPushButton>
 #include <QTimer>
 
-PandaStream::PandaStream(QObject *parent, PandaStreamConfig config_) : config(config_), LiveStream(parent) {
+PandaStream::PandaStream(PandaStreamConfig config_) : config(config_) {
   if (!connect()) {
     throw std::runtime_error("Failed to connect to panda");
   }
@@ -181,7 +181,7 @@ void OpenPandaWidget::buildConfigForm() {
 
 AbstractStream *OpenPandaWidget::open() {
   try {
-    return new PandaStream(qApp, config);
+    return new PandaStream(config);
   } catch (std::exception &e) {
     QMessageBox::warning(nullptr, tr("Warning"), tr("Failed to connect to panda: '%1'").arg(e.what()));
     return nullptr;
diff --git a/openpilot/tools/cabana/streams/pandastream.h b/openpilot/tools/cabana/streams/pandastream.h
index f8847f65e..d764613fe 100644
--- a/openpilot/tools/cabana/streams/pandastream.h
+++ b/openpilot/tools/cabana/streams/pandastream.h
@@ -24,9 +24,8 @@ struct PandaStreamConfig {
 };
 
 class PandaStream : public LiveStream {
-  Q_OBJECT
 public:
-  PandaStream(QObject *parent, PandaStreamConfig config_ = {});
+  PandaStream(PandaStreamConfig config_ = {});
   ~PandaStream() { stop(); }
   inline std::string routeName() const override {
     return "Panda: " + config.serial;
diff --git a/openpilot/tools/cabana/streams/replaystream.cc b/openpilot/tools/cabana/streams/replaystream.cc
index 6d54369ae..68f4fb305 100644
--- a/openpilot/tools/cabana/streams/replaystream.cc
+++ b/openpilot/tools/cabana/streams/replaystream.cc
@@ -12,17 +12,21 @@
 #include "common/util.h"
 #include "tools/cabana/streams/routes.h"
 
-ReplayStream::ReplayStream(QObject *parent) : AbstractStream(parent) {
+ReplayStream::ReplayStream() {
   unsetenv("ZMQ");
   setenv("COMMA_CACHE", "/tmp/comma_download_cache", 1);
 
   op_prefix = std::make_unique<OpenpilotPrefix>();
 
-  QObject::connect(&settings, &Settings::changed, this, [this]() {
+  settings_connection_ = settings.changed.connect([this]() {
     if (replay) replay->setSegmentCacheLimit(settings.max_cached_minutes);
   });
 }
 
+ReplayStream::~ReplayStream() {
+  cancelWaits();
+}
+
 void ReplayStream::mergeSegments() {
   auto event_data = replay->getEventData();
   for (const auto &[n, seg] : event_data->segments) {
@@ -51,14 +55,14 @@ bool ReplayStream::loadRoute(const std::string &route, const std::string &data_d
   replay->setSegmentCacheLimit(settings.max_cached_minutes);
   replay->installEventFilter([this](const Event *event) { return eventFilter(event); });
 
-  // Forward replay callbacks to corresponding Qt signals.
-  replay->onSeeking = [this](double sec) { emit seeking(sec); };
+  // replay callbacks arrive on replay threads
+  replay->onSeeking = [this](double sec) { postToMainThread([this, sec]() { seeking(sec); }); };
   replay->onSeekedTo = [this](double sec) {
-    emit seekedTo(sec);
+    postToMainThread([this, sec]() { seekedTo(sec); });
     waitForSeekFinshed();
   };
-  replay->onQLogLoaded = [this](std::shared_ptr<LogReader> qlog) { emit qLogLoaded(qlog); };
-  replay->onSegmentsMerged = [this]() { QMetaObject::invokeMethod(this, &ReplayStream::mergeSegments, Qt::BlockingQueuedConnection); };
+  replay->onQLogLoaded = [this](std::shared_ptr<LogReader> qlog) { postToMainThread([this, qlog]() { qLogLoaded(qlog); }); };
+  replay->onSegmentsMerged = [this]() { postToMainThreadAndWait([this]() { mergeSegments(); }); };
 
   bool success = replay->load();
   if (!success) {
@@ -70,18 +74,18 @@ bool ReplayStream::loadRoute(const std::string &route, const std::string &data_d
                   "python3 openpilot/tools/lib/auth.py\n\n"
                   "This will grant access to routes from your comma account.";
       } else {
-        message = tr("Access Denied. You do not have permission to access route:\n\n%1\n\n"
-                     "This is likely a private route.").arg(QString::fromStdString(route));
+        message = QString("Access Denied. You do not have permission to access route:\n\n%1\n\n"
+                          "This is likely a private route.").arg(QString::fromStdString(route));
       }
-      QMessageBox::warning(nullptr, tr("Access Denied"), message);
+      QMessageBox::warning(nullptr, "Access Denied", message);
     } else if (replay->lastRouteError() == RouteLoadError::NetworkError) {
-      QMessageBox::warning(nullptr, tr("Network Error"),
-                          tr("Unable to load the route:\n\n %1.\n\nPlease check your network connection and try again.").arg(QString::fromStdString(route)));
+      QMessageBox::warning(nullptr, "Network Error",
+                          QString("Unable to load the route:\n\n %1.\n\nPlease check your network connection and try again.").arg(QString::fromStdString(route)));
     } else if (replay->lastRouteError() == RouteLoadError::FileNotFound) {
-      QMessageBox::warning(nullptr, tr("Route Not Found"),
-                           tr("The specified route could not be found:\n\n %1.\n\nPlease check the route name and try again.").arg(QString::fromStdString(route)));
+      QMessageBox::warning(nullptr, "Route Not Found",
+                           QString("The specified route could not be found:\n\n %1.\n\nPlease check the route name and try again.").arg(QString::fromStdString(route)));
     } else {
-      QMessageBox::warning(nullptr, tr("Route Load Failed"), tr("Failed to load route: '%1'").arg(QString::fromStdString(route)));
+      QMessageBox::warning(nullptr, "Route Load Failed", QString("Failed to load route: '%1'").arg(QString::fromStdString(route)));
     }
   }
   return success;
@@ -102,7 +106,7 @@ bool ReplayStream::eventFilter(const Event *event) {
 
   double ts = millis_since_boot();
   if ((ts - prev_update_ts) > (1000.0 / settings.fps)) {
-    emit privateUpdateLastMsgsSignal();
+    requestUpdateLastMessages();
     prev_update_ts = ts;
   }
   return true;
@@ -110,7 +114,7 @@ bool ReplayStream::eventFilter(const Event *event) {
 
 void ReplayStream::pause(bool pause) {
   replay->pause(pause);
-  emit(pause ? paused() : resume());
+  pause ? paused() : resume();
 }
 
 
@@ -161,7 +165,7 @@ AbstractStream *OpenReplayWidget::open() {
   if (!is_valid_format) {
     QMessageBox::warning(nullptr, tr("Warning"), tr("Invalid route format: '%1'").arg(route));
   } else {
-    auto replay_stream = std::make_unique<ReplayStream>(qApp);
+    auto replay_stream = std::make_unique<ReplayStream>();
     uint32_t flags = REPLAY_FLAG_NONE;
     if (cameras[1]->isChecked()) flags |= REPLAY_FLAG_CABIN_CAMERA;
     if (cameras[2]->isChecked()) flags |= REPLAY_FLAG_WIDE_ROAD;
diff --git a/openpilot/tools/cabana/streams/replaystream.h b/openpilot/tools/cabana/streams/replaystream.h
index eecd34571..6f49502ba 100644
--- a/openpilot/tools/cabana/streams/replaystream.h
+++ b/openpilot/tools/cabana/streams/replaystream.h
@@ -10,13 +10,10 @@
 #include "tools/cabana/streams/abstractstream.h"
 #include "tools/replay/replay.h"
 
-Q_DECLARE_METATYPE(std::shared_ptr<LogReader>);
-
 class ReplayStream : public AbstractStream {
-  Q_OBJECT
-
 public:
-  ReplayStream(QObject *parent);
+  ReplayStream();
+  ~ReplayStream();
   void start() override { replay->start(); }
   bool loadRoute(const std::string &route, const std::string &data_dir, uint32_t replay_flags = REPLAY_FLAG_NONE, bool auto_source = false);
   bool eventFilter(const Event *event);
@@ -36,12 +33,13 @@ public:
   inline bool isPaused() const override { return replay->isPaused(); }
   void pause(bool pause) override;
 
-signals:
-  void qLogLoaded(std::shared_ptr<LogReader> qlog);
+  // invoked on the main thread
+  Observable<std::shared_ptr<LogReader>> qLogLoaded;
 
 private:
   void mergeSegments();
   std::unique_ptr<Replay> replay = nullptr;
+  Connection settings_connection_;
   std::set<int> processed_segments;
   std::unique_ptr<OpenpilotPrefix> op_prefix;
 };
diff --git a/openpilot/tools/cabana/streams/socketcanstream.cc b/openpilot/tools/cabana/streams/socketcanstream.cc
index b616e7f24..b10f2cab9 100644
--- a/openpilot/tools/cabana/streams/socketcanstream.cc
+++ b/openpilot/tools/cabana/streams/socketcanstream.cc
@@ -16,7 +16,7 @@
 #include <QMessageBox>
 #include <QPushButton>
 
-SocketCanStream::SocketCanStream(QObject *parent, SocketCanStreamConfig config_) : config(config_), LiveStream(parent) {
+SocketCanStream::SocketCanStream(SocketCanStreamConfig config_) : config(config_) {
   if (!available()) {
     throw std::runtime_error("SocketCAN not available");
   }
@@ -140,7 +140,7 @@ void OpenSocketCanWidget::refreshDevices() {
 
 AbstractStream *OpenSocketCanWidget::open() {
   try {
-    return new SocketCanStream(qApp, config);
+    return new SocketCanStream(config);
   } catch (std::exception &e) {
     QMessageBox::warning(nullptr, tr("Warning"), tr("Failed to connect to SocketCAN device: '%1'").arg(e.what()));
     return nullptr;
diff --git a/openpilot/tools/cabana/streams/socketcanstream.h b/openpilot/tools/cabana/streams/socketcanstream.h
index 3c5cd184f..78e856d50 100644
--- a/openpilot/tools/cabana/streams/socketcanstream.h
+++ b/openpilot/tools/cabana/streams/socketcanstream.h
@@ -9,9 +9,8 @@ struct SocketCanStreamConfig {
 };
 
 class SocketCanStream : public LiveStream {
-  Q_OBJECT
 public:
-  SocketCanStream(QObject *parent, SocketCanStreamConfig config_ = {});
+  SocketCanStream(SocketCanStreamConfig config_ = {});
   ~SocketCanStream();
   static bool available();
 
diff --git a/openpilot/tools/cabana/tests/test_cabana.cc b/openpilot/tools/cabana/tests/test_cabana.cc
index 53be1b0af..c2ebc2e6b 100644
--- a/openpilot/tools/cabana/tests/test_cabana.cc
+++ b/openpilot/tools/cabana/tests/test_cabana.cc
@@ -164,11 +164,10 @@ void test_dbc_manager() {
   int files_changed = 0;
   int signals_added = 0;
   int masks_updated = 0;
-  manager.setCallbacks({
-    .signal_added = [&](MessageId, const cabana::Signal *) { ++signals_added; },
-    .file_changed = [&]() { ++files_changed; },
-    .mask_updated = [&]() { ++masks_updated; },
-  });
+  Connections connections;
+  connections.push_back(manager.signalAdded.connect([&](MessageId, const cabana::Signal *) { ++signals_added; }));
+  connections.push_back(manager.fileChanged.connect([&]() { ++files_changed; }));
+  connections.push_back(manager.maskUpdated.connect([&]() { ++masks_updated; }));
 
   std::string error;
   REQUIRE(manager.open(SOURCE_ALL, "test", "BO_ 160 message: 8 XXX\n", &error));
diff --git a/openpilot/tools/cabana/tools/routeinfo.cc b/openpilot/tools/cabana/tools/routeinfo.cc
index dc272e3d1..1037d4a20 100644
--- a/openpilot/tools/cabana/tools/routeinfo.cc
+++ b/openpilot/tools/cabana/tools/routeinfo.cc
@@ -6,7 +6,7 @@
 #include "tools/cabana/streams/replaystream.h"
 
 RouteInfoDlg::RouteInfoDlg(QWidget *parent) : QDialog(parent) {
-  auto *replay = qobject_cast<ReplayStream *>(can)->getReplay();
+  auto *replay = dynamic_cast<ReplayStream *>(can)->getReplay();
   setWindowTitle(tr("Route: %1").arg(QString::fromStdString(replay->route().name())));
 
   auto *table = new QTableWidget(replay->route().segments().size(), 7, this);
diff --git a/openpilot/tools/cabana/utils/util.cc b/openpilot/tools/cabana/utils/util.cc
index 87a0f8942..9a26e5f97 100644
--- a/openpilot/tools/cabana/utils/util.cc
+++ b/openpilot/tools/cabana/utils/util.cc
@@ -11,7 +11,9 @@
 #include <filesystem>
 #include <limits>
 #include <memory>
+#include <mutex>
 #include <string>
+#include <thread>
 #include <sys/socket.h>
 #include <sys/wait.h>
 #include <unistd.h>
@@ -23,6 +25,30 @@
 #include <unordered_map>
 #include "common/util.h"
 
+static const std::thread::id main_thread_id = std::this_thread::get_id();
+static std::mutex main_thread_queue_mutex;
+static std::vector<std::function<void()>> main_thread_queue;
+
+bool utils::isMainThread() { return std::this_thread::get_id() == main_thread_id; }
+
+void utils::runOnMainThread(std::function<void()> fn) {
+  if (isMainThread()) {
+    fn();
+  } else {
+    std::lock_guard lk(main_thread_queue_mutex);
+    main_thread_queue.push_back(std::move(fn));
+  }
+}
+
+void utils::drainMainThreadQueue() {
+  std::vector<std::function<void()>> fns;
+  {
+    std::lock_guard lk(main_thread_queue_mutex);
+    fns.swap(main_thread_queue);
+  }
+  for (auto &fn : fns) fn();
+}
+
 // SegmentTree
 
 void SegmentTree::build(const std::vector<QPointF> &arr) {
diff --git a/openpilot/tools/cabana/utils/util.h b/openpilot/tools/cabana/utils/util.h
index 5a3c62d11..cbcb6d14e 100644
--- a/openpilot/tools/cabana/utils/util.h
+++ b/openpilot/tools/cabana/utils/util.h
@@ -4,6 +4,7 @@
 #include <atomic>
 #include <cmath>
 #include <filesystem>
+#include <functional>
 #include <string>
 #include <thread>
 #include <vector>
@@ -20,9 +21,14 @@
 #include <QToolButton>
 #include <QValidator>
 
+#include "tools/cabana/core/observable.h"
 #include "tools/cabana/dbc/dbc.h"
 #include "tools/cabana/settings.h"
 
+// needed by QVariant::fromValue() in the Qt views; goes away with QVariant
+Q_DECLARE_METATYPE(MessageId)
+Q_DECLARE_METATYPE(ValueDescription)
+
 inline QColor toQColor(const CabanaColor &color) {
   return QColor(color.r, color.g, color.b, color.a);
 }
@@ -132,6 +138,10 @@ public:
 
 namespace utils {
 
+bool isMainThread();
+// inline on the main thread, queued until drainMainThreadQueue() otherwise
+void runOnMainThread(std::function<void()> fn);
+void drainMainThreadQueue();
 QPixmap icon(const QString &id);
 std::string homePath();
 std::filesystem::path configPath();
@@ -175,7 +185,7 @@ public:
     const int metric = QApplication::style()->pixelMetric(QStyle::PM_SmallIconSize);
     setIconSize({metric, metric});
     theme = settings.theme;
-    connect(&settings, &Settings::changed, this, &ToolButton::updateIcon);
+    settings_connection_ = settings.changed.connect([this]() { updateIcon(); });
   }
   void setIcon(const QString &icon) {
     icon_str = icon;
@@ -184,6 +194,7 @@ public:
 
 private:
   void updateIcon() { if (std::exchange(theme, settings.theme) != theme) setIcon(icon_str); }
+  Connection settings_connection_;
   QString icon_str;
   int theme;
 };
diff --git a/openpilot/tools/cabana/videowidget.cc b/openpilot/tools/cabana/videowidget.cc
index bd6157365..b1e84586a 100644
--- a/openpilot/tools/cabana/videowidget.cc
+++ b/openpilot/tools/cabana/videowidget.cc
@@ -27,7 +27,7 @@ static const QColor timeline_colors[] = {
 };
 
 static Replay *getReplay() {
-  auto stream = qobject_cast<ReplayStream *>(can);
+  auto stream = dynamic_cast<ReplayStream *>(can);
   return stream ? stream->getReplay() : nullptr;
 }
 
@@ -42,11 +42,11 @@ VideoWidget::VideoWidget(QWidget *parent) : QFrame(parent) {
   createPlaybackController();
 
   setSizePolicy(QSizePolicy::Preferred, QSizePolicy::Maximum);
-  QObject::connect(can, &AbstractStream::paused, this, &VideoWidget::updatePlayBtnState);
-  QObject::connect(can, &AbstractStream::resume, this, &VideoWidget::updatePlayBtnState);
-  QObject::connect(can, &AbstractStream::msgsReceived, this, &VideoWidget::updateState);
-  QObject::connect(can, &AbstractStream::seeking, this, &VideoWidget::updateState);
-  QObject::connect(can, &AbstractStream::timeRangeChanged, this, &VideoWidget::timeRangeChanged);
+  connections_.push_back(can->paused.connect([this]() { updatePlayBtnState(); }));
+  connections_.push_back(can->resume.connect([this]() { updatePlayBtnState(); }));
+  connections_.push_back(can->msgsReceived.connect([this](const std::set<MessageId> *, bool) { updateState(); }));
+  connections_.push_back(can->seeking.connect([this](double) { updateState(); }));
+  connections_.push_back(can->timeRangeChanged.connect([this](const auto &) { timeRangeChanged(); }));
 
   updatePlayBtnState();
   setWhatsThis(tr(R"(
@@ -157,14 +157,14 @@ QWidget *VideoWidget::createCameraWidget() {
   slider->setTimeRange(can->minSeconds(), can->maxSeconds());
 
   QObject::connect(slider, &QSlider::sliderReleased, [this]() { can->seekTo(slider->currentSecond()); });
-  QObject::connect(can, &AbstractStream::paused, cam_widget, qOverload<>(&StreamCameraView::update));
-  QObject::connect(can, &AbstractStream::eventsMerged, this, [this]() { slider->update(); });
-  QObject::connect(cam_widget, &CameraWidget::clicked, []() { can->pause(!can->isPaused()); });
-  QObject::connect(cam_widget, &CameraWidget::vipcAvailableStreamsUpdated, this, &VideoWidget::vipcAvailableStreamsUpdated);
+  connections_.push_back(can->paused.connect([this]() { cam_widget->update(); }));
+  connections_.push_back(can->eventsMerged.connect([this](const MessageEventsMap &) { slider->update(); }));
+  connections_.push_back(cam_widget->clicked.connect([]() { can->pause(!can->isPaused()); }));
+  connections_.push_back(cam_widget->availableStreamsUpdated.connect([this](std::set<VisionStreamType> streams) { vipcAvailableStreamsUpdated(streams); }));
   QObject::connect(camera_tab, &QTabBar::currentChanged, [this](int index) {
     if (index != -1) cam_widget->setStreamType((VisionStreamType)camera_tab->tabData(index).toInt());
   });
-  QObject::connect(static_cast<ReplayStream*>(can), &ReplayStream::qLogLoaded, cam_widget, &StreamCameraView::parseQLog, Qt::QueuedConnection);
+  connections_.push_back(static_cast<ReplayStream *>(can)->qLogLoaded.connect([this](std::shared_ptr<LogReader> qlog) { cam_widget->parseQLog(qlog); }));
   slider->installEventFilter(this);
   return w;
 }
diff --git a/openpilot/tools/cabana/videowidget.h b/openpilot/tools/cabana/videowidget.h
index 09f7a9931..1b6d81406 100644
--- a/openpilot/tools/cabana/videowidget.h
+++ b/openpilot/tools/cabana/videowidget.h
@@ -64,6 +64,7 @@ protected:
   void timeRangeChanged();
   void updateState();
   void updatePlayBtnState();
+  Connections connections_;
   QWidget *createCameraWidget();
   void createPlaybackController();
   void createSpeedDropdown(QToolBar *toolbar);