diff --git a/PROJECT_PLAN.md b/PROJECT_PLAN.md
index ea40863..42001f0 100644
--- a/PROJECT_PLAN.md
+++ b/PROJECT_PLAN.md
@@ -1197,7 +1197,7 @@ private:
 - [x] Integrate libpipewire with `pw_thread_loop`
 - [x] Implement node/port discovery via registry callbacks
 - [x] Implement link creation/destruction
-- [ ] Create lock-free communication primitives (atomics, ring buffers)
+- [x] Create lock-free communication primitives (atomics, ring buffers)
 - [x] **Acceptance Criteria:** CLI test app that lists nodes and creates a link programmatically
 
 ### Milestone 2: QtNodes Integration
@@ -1205,8 +1205,8 @@ private:
 - [x] Integrate QtNodes library (submodule or CMake package)
 - [x] Create `AudioNodeDataModel` for PipeWire nodes
 - [x] Map PipeWire ports to QtNodes handles
-- [ ] Implement connection validation
-- [ ] Create custom node widgets with embedded controls
+- [x] Implement connection validation
+- [x] Create custom node widgets with embedded controls
 - [x] **Acceptance Criteria:** Visual graph editor displays PipeWire nodes and allows dragging connections
 
 ### Milestone 3: Real-Time Meters & Performance
diff --git a/src/gui/PipeWireGraphModel.cpp b/src/gui/PipeWireGraphModel.cpp
index 06c1455..9b4e811 100644
--- a/src/gui/PipeWireGraphModel.cpp
+++ b/src/gui/PipeWireGraphModel.cpp
@@ -16,6 +16,11 @@
 #include <QDir>
 #include <QFile>
 #include <QFileInfo>
+#include <QHBoxLayout>
+#include <QSlider>
+#include <QSizePolicy>
+#include <QToolButton>
+#include <QWidget>
 
 #include <QtNodes/NodeStyle>
 #include <QtNodes/StyleCollection>
@@ -108,6 +113,51 @@ PipeWireGraphModel::PipeWireGraphModel(Potato::PipeWireController *controller, Q
     }
 }
 
+QWidget *PipeWireGraphModel::nodeWidget(QtNodes::NodeId nodeId) const
+{
+    auto it = m_nodeWidgets.find(nodeId);
+    if (it != m_nodeWidgets.end()) {
+        return it->second;
+    }
+
+    auto *widget = new QWidget();
+    auto *layout = new QHBoxLayout(widget);
+    layout->setContentsMargins(6, 2, 6, 2);
+    layout->setSpacing(6);
+
+    auto *muteButton = new QToolButton(widget);
+    muteButton->setText("M");
+    muteButton->setCheckable(true);
+    muteButton->setFixedSize(20, 20);
+    muteButton->setToolTip(QString("Mute (UI only)"));
+
+    auto *slider = new QSlider(Qt::Horizontal, widget);
+    slider->setRange(0, 100);
+    slider->setValue(100);
+    slider->setFixedHeight(18);
+    slider->setSizePolicy(QSizePolicy::Expanding, QSizePolicy::Fixed);
+    slider->setToolTip(QString("Volume (UI only)"));
+
+    layout->addWidget(muteButton);
+    layout->addWidget(slider);
+
+    widget->setSizePolicy(QSizePolicy::Preferred, QSizePolicy::Fixed);
+    widget->setFixedHeight(26);
+    widget->setStyleSheet(
+        "QToolButton { background: #2b2f38; color: #dbe2ee; border: 1px solid #39404c; border-radius: 4px; font-weight: 700; }"
+        "QToolButton:checked { background: #3c4350; color: #ffcf7a; }"
+        "QSlider::groove:horizontal { height: 4px; background: #2b2f38; border-radius: 2px; }"
+        "QSlider::sub-page:horizontal { background: #5fcf8d; border-radius: 2px; }"
+        "QSlider::add-page:horizontal { background: #2b2f38; border-radius: 2px; }"
+        "QSlider::handle:horizontal { width: 10px; margin: -6px 0; background: #dbe2ee; border-radius: 5px; }"
+    );
+
+    widget->adjustSize();
+
+    m_nodeWidgets.emplace(nodeId, widget);
+    return widget;
+}
+
 QtNodes::NodeId PipeWireGraphModel::newNodeId()
 {
     return m_nextNodeId++;
@@ -210,6 +260,16 @@ bool PipeWireGraphModel::connectionPossible(QtNodes::ConnectionId const connecti
         return false;
     }
 
+    const auto &outPort = outInfo.outputPorts.at(connectionId.outPortIndex);
+    const auto &inPort = inInfo.inputPorts.at(connectionId.inPortIndex);
+    if (outPort.direction != SPA_DIRECTION_OUTPUT || inPort.direction != SPA_DIRECTION_INPUT) {
+        return false;
+    }
+
+    if (outInfo.mediaClass != inInfo.mediaClass) {
+        return false;
+    }
+
     return true;
 }
 
@@ -257,6 +317,10 @@ QVariant PipeWireGraphModel::nodeData(QtNodes::NodeId nodeId, QtNodes::NodeRole
         }
         case QtNodes::NodeRole::Size:
         {
+            auto sizeIt = m_nodeSizes.find(nodeId);
+            if (sizeIt != m_nodeSizes.end()) {
+                return sizeIt->second;
+            }
             const int maxPorts = std::max(info.inputPorts.size(), info.outputPorts.size());
             const int baseHeight = 50;
             const int perPortHeight = 28;
@@ -274,6 +338,8 @@ QVariant PipeWireGraphModel::nodeData(QtNodes::NodeId nodeId, QtNodes::NodeRole
             return QString("PipeWire");
         case QtNodes::NodeRole::Style:
             return nodeStyleVariant(info);
+        case QtNodes::NodeRole::Widget:
+            return QVariant::fromValue(nodeWidget(nodeId));
         default:
             return QVariant();
     }
@@ -293,6 +359,11 @@ bool PipeWireGraphModel::setNodeData(QtNodes::NodeId nodeId, QtNodes::NodeRole r
         return true;
     }
 
+    if (role == QtNodes::NodeRole::Size) {
+        m_nodeSizes[nodeId] = value.toSize();
+        return true;
+    }
+
     return false;
 }
 
@@ -333,9 +404,6 @@ QVariant PipeWireGraphModel::portData(QtNodes::NodeId nodeId,
     }
 
     if (role == QtNodes::PortRole::ConnectionPolicyRole) {
-        if (portType == QtNodes::PortType::In) {
-            return QVariant::fromValue(QtNodes::ConnectionPolicy::One);
-        }
         return QVariant::fromValue(QtNodes::ConnectionPolicy::Many);
     }
 
@@ -379,6 +447,8 @@ bool PipeWireGraphModel::deleteNode(QtNodes::NodeId const nodeId)
 
     m_nodes.erase(nodeId);
     m_positions.erase(nodeId);
+    m_nodeSizes.erase(nodeId);
+    m_nodeWidgets.erase(nodeId);
     Q_EMIT nodeDeleted(nodeId);
     return true;
 }
@@ -476,6 +546,7 @@ bool PipeWireGraphModel::updatePipeWireNode(const Potato::NodeInfo &node)
 
     const QtNodes::NodeId nodeId = it->second;
     m_nodes[nodeId] = node;
+    m_nodeSizes.erase(nodeId);
     Q_EMIT nodeUpdated(nodeId);
     return true;
 }
@@ -503,6 +574,8 @@ void PipeWireGraphModel::reset()
     m_nodes.clear();
     m_pwToNode.clear();
     m_positions.clear();
+    m_nodeSizes.clear();
+    m_nodeWidgets.clear();
     m_nextNodeId = 1;
     Q_EMIT modelReset();
 }
diff --git a/src/gui/PipeWireGraphModel.h b/src/gui/PipeWireGraphModel.h
index 891df31..64ef775 100644
--- a/src/gui/PipeWireGraphModel.h
+++ b/src/gui/PipeWireGraphModel.h
@@ -16,6 +16,8 @@
 #include <unordered_map>
 #include <unordered_set>
 
+class QWidget;
+
 class PipeWireGraphModel : public QtNodes::AbstractGraphModel
 {
     Q_OBJECT
@@ -72,6 +74,7 @@ public:
     bool splitterSizes(QList<int> &sizes) const;
 
 private:
+    QWidget *nodeWidget(QtNodes::NodeId nodeId) const;
     QtNodes::ConnectionId connectionFromPipeWire(const Potato::LinkInfo &link, bool *ok) const;
     bool findPortIndex(const Potato::NodeInfo &node, uint32_t portId, QtNodes::PortType type, QtNodes::PortIndex &index) const;
     QString portLabel(const Potato::PortInfo &port) const;
@@ -95,4 +98,6 @@ private:
     bool m_hasViewState = false;
     QList<int> m_splitterSizes;
     bool m_hasSplitterSizes = false;
+    mutable std::unordered_map<QtNodes::NodeId, QWidget*> m_nodeWidgets;
+    std::unordered_map<QtNodes::NodeId, QSize> m_nodeSizes;
 };
diff --git a/src/pipewire/pipewirecontroller.cpp b/src/pipewire/pipewirecontroller.cpp
index 9108939..f1b305e 100644
--- a/src/pipewire/pipewirecontroller.cpp
+++ b/src/pipewire/pipewirecontroller.cpp
@@ -21,6 +21,59 @@
 
 namespace Potato {
 
+static constexpr uint32_t kMeterRingCapacityBytes = 4096;
+
+static void writeRingValue(spa_ringbuffer *ring, std::vector<uint8_t> &buffer, float value)
+{
+    if (!ring || buffer.empty()) {
+        return;
+    }
+
+    uint32_t index = 0;
+    const uint32_t avail = spa_ringbuffer_get_write_index(ring, &index);
+    if (avail < sizeof(float)) {
+        return;
+    }
+
+    const uint32_t size = static_cast<uint32_t>(buffer.size());
+    uint32_t offset = index & (size - 1);
+    if (offset + sizeof(float) <= size) {
+        std::memcpy(buffer.data() + offset, &value, sizeof(float));
+    } else {
+        const uint32_t first = size - offset;
+        std::memcpy(buffer.data() + offset, &value, first);
+        std::memcpy(buffer.data(), reinterpret_cast<const uint8_t*>(&value) + first, sizeof(float) - first);
+    }
+    spa_ringbuffer_write_update(ring, index + sizeof(float));
+}
+
+static bool readRingLatest(spa_ringbuffer *ring, std::vector<uint8_t> &buffer, float &value)
+{
+    if (!ring || buffer.empty()) {
+        return false;
+    }
+
+    uint32_t index = 0;
+    const uint32_t avail = spa_ringbuffer_get_read_index(ring, &index);
+    if (avail < sizeof(float)) {
+        return false;
+    }
+
+    const uint32_t size = static_cast<uint32_t>(buffer.size());
+    const uint32_t latestIndex = index + (avail - sizeof(float));
+    uint32_t offset = latestIndex & (size - 1);
+    if (offset + sizeof(float) <= size) {
+        std::memcpy(&value, buffer.data() + offset, sizeof(float));
+    } else {
+        const uint32_t first = size - offset;
+        std::memcpy(&value, buffer.data() + offset, first);
+        std::memcpy(reinterpret_cast<uint8_t*>(&value) + first, buffer.data(), sizeof(float) - first);
+    }
+
+    spa_ringbuffer_read_update(ring, index + avail);
+    return true;
+}
+
 static QString toQString(const char *value)
 {
     if (!value) {
@@ -150,6 +203,9 @@ void meterProcess(void *data)
     }
 
     self->m_meterPeak.store(peak, std::memory_order_relaxed);
+    if (self->m_meterRingReady.load(std::memory_order_relaxed)) {
+        writeRingValue(&self->m_meterRing, self->m_meterRingData, peak);
+    }
     pw_stream_queue_buffer(self->m_meterStream, buf);
 }
 
@@ -217,6 +273,9 @@ PipeWireController::PipeWireController(QObject *parent)
 {
     m_registryListener = new spa_hook;
     m_coreListener = new spa_hook;
+    m_meterRingData.resize(kMeterRingCapacityBytes);
+    spa_ringbuffer_init(&m_meterRing);
+    m_meterRingReady.store(true, std::memory_order_relaxed);
 }
 
 PipeWireController::~PipeWireController()
@@ -340,6 +399,9 @@ void PipeWireController::shutdown()
     }
     
     pw_deinit();
+
+    m_meterRingReady.store(false, std::memory_order_relaxed);
+    m_meterRingData.clear();
     
     m_initialized.storeRelaxed(false);
     m_connected.storeRelaxed(false);
@@ -372,7 +434,14 @@ QVector<LinkInfo> PipeWireController::links() const
 
 float PipeWireController::meterPeak() const
 {
-    return m_meterPeak.load(std::memory_order_relaxed);
+    float peak = m_meterPeak.load(std::memory_order_relaxed);
+    if (m_meterRingReady.load(std::memory_order_relaxed)) {
+        float ringPeak = 0.0f;
+        if (readRingLatest(&m_meterRing, m_meterRingData, ringPeak)) {
+            peak = ringPeak;
+        }
+    }
+    return peak;
 }
 
 float PipeWireController::nodeMeterPeak(uint32_t nodeId) const
diff --git a/src/pipewire/pipewirecontroller.h b/src/pipewire/pipewirecontroller.h
index 322f95e..1d9444d 100644
--- a/src/pipewire/pipewirecontroller.h
+++ b/src/pipewire/pipewirecontroller.h
@@ -6,6 +6,9 @@
 #include <QMutex>
 #include <QAtomicInteger>
 #include <atomic>
+#include <vector>
+
+#include <spa/utils/ringbuffer.h>
 
 struct pw_thread_loop;
 struct pw_context;
@@ -93,6 +96,9 @@ private:
     QAtomicInteger<bool> m_connected{false};
     QAtomicInteger<bool> m_initialized{false};
     std::atomic<float> m_meterPeak{0.0f};
+    mutable spa_ringbuffer m_meterRing{};
+    mutable std::vector<uint8_t> m_meterRingData;
+    std::atomic<bool> m_meterRingReady{false};
 
     mutable QMutex m_meterMutex;
     QMap<uint32_t, NodeMeter*> m_nodeMeters;