feat: real-time dB level meters per node

Backend:
- One pw_stream (INPUT, F32) per audio node; for sinks use
  stream.capture.sink=true to read the monitor port
- RT process callback computes instantaneous linear peak with no
  allocations; stored in std::atomic<float>
- Meter streams are filtered from the registry so they never appear
  in the graph or trigger recursive meter creation
- Meter streams are created on first node-ready event, destroyed on
  node removal, and cleaned up on engine close
- GET /api/peaks → {node_id: linear_peak} (polled by frontend)

Frontend:
- peaks store polled at 100 ms via setInterval; starts/stops with
  initGraph/destroyGraph
- Each node card grows 8 px and shows a 3 px meter bar at the bottom
  (green below -12 dB, yellow -12 to -3 dB, red above -3 dB)

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

src/graph_engine.cpp

@@ -27,6 +27,138 @@
 
 using namespace pwgraph;
 
+// ============================================================================
+// Level meter: pw_stream per audio node, RT process callback stores peak
+// ============================================================================
+
+GraphEngine::MeterStream::~MeterStream() {
+    if (stream) {
+        pw_stream_destroy(stream);
+        stream = nullptr;
+    }
+}
+
+static void on_meter_process(void *data) {
+    auto *ms = static_cast<GraphEngine::MeterStream*>(data);
+    struct pw_buffer *buf = pw_stream_dequeue_buffer(ms->stream);
+    if (!buf) return;
+
+    float peak = 0.0f;
+    struct spa_buffer *sbuf = buf->buffer;
+    for (uint32_t d = 0; d < sbuf->n_datas; d++) {
+        if (!sbuf->datas[d].data) continue;
+        const float *s = static_cast<const float*>(sbuf->datas[d].data);
+        uint32_t n = sbuf->datas[d].chunk->size / sizeof(float);
+        for (uint32_t i = 0; i < n; i++) {
+            float v = s[i] < 0 ? -s[i] : s[i];
+            if (v > peak) peak = v;
+        }
+    }
+    ms->peak.store(peak, std::memory_order_relaxed);
+    pw_stream_queue_buffer(ms->stream, buf);
+}
+
+static const struct pw_stream_events meter_stream_events = {
+    .version = PW_VERSION_STREAM_EVENTS,
+    .process = on_meter_process,
+};
+
+void GraphEngine::createMeterIfNeeded(uint32_t node_id, PortMode mode,
+                                      NodeType ntype, const std::string& name)
+{
+    // Only meter audio nodes
+    if ((ntype & NodeType::Audio) == NodeType::None) return;
+    // Skip nodes with no meaningful direction
+    if (mode == PortMode::None) return;
+
+    {
+        std::lock_guard<std::mutex> lk(m_meter_mutex);
+        if (m_meters.count(node_id)) return;  // already created
+    }
+
+    // Sinks (speakers/headphones) are metered via their monitor port
+    bool capture_sink = (mode == PortMode::Input || mode == PortMode::Duplex);
+
+    std::string meter_node_name = "pwweb.meter." + std::to_string(node_id);
+
+    struct pw_properties *props = pw_properties_new(
+        PW_KEY_MEDIA_TYPE,       "Audio",
+        PW_KEY_MEDIA_CATEGORY,   "Capture",
+        PW_KEY_MEDIA_ROLE,       "DSP",
+        PW_KEY_APP_NAME,         "pwweb-meter",
+        PW_KEY_NODE_NAME,        meter_node_name.c_str(),
+        PW_KEY_TARGET_OBJECT,    std::to_string(node_id).c_str(),
+        "stream.capture.sink",   capture_sink ? "true" : "false",
+        nullptr);
+
+    struct pw_stream *stream = pw_stream_new(m_pw.core, "pwweb-meter", props);
+    if (!stream) {
+        fprintf(stderr, "pwweb: meter stream alloc failed for node %u\n", node_id);
+        return;
+    }
+
+    auto *ms       = new MeterStream();
+    ms->node_id    = node_id;
+    ms->stream     = stream;
+    ms->engine_ref = this;
+
+    pw_stream_add_listener(stream, &ms->listener, &meter_stream_events, ms);
+
+    uint8_t pod_buf[1024];
+    struct spa_pod_builder b = SPA_POD_BUILDER_INIT(pod_buf, sizeof(pod_buf));
+    const struct spa_pod *params[1];
+    struct spa_audio_info_raw info = {};
+    info.format   = SPA_AUDIO_FORMAT_F32;
+    info.rate     = 48000;
+    info.channels = 2;
+    params[0] = spa_format_audio_raw_build(&b, SPA_PARAM_EnumFormat, &info);
+
+    int ret = pw_stream_connect(stream,
+        PW_DIRECTION_INPUT,
+        SPA_ID_INVALID,
+        static_cast<pw_stream_flags>(PW_STREAM_FLAG_AUTOCONNECT |
+                                     PW_STREAM_FLAG_MAP_BUFFERS),
+        params, 1);
+
+    if (ret < 0) {
+        fprintf(stderr, "pwweb: meter connect failed node %u (%s): %s\n",
+                node_id, name.c_str(), spa_strerror(ret));
+        delete ms;
+        return;
+    }
+
+    {
+        std::lock_guard<std::mutex> lk(m_meter_mutex);
+        m_meters[node_id] = ms;
+    }
+    fprintf(stderr, "pwweb: meter created for node %u (%s%s)\n",
+            node_id, name.c_str(), capture_sink ? ", monitor" : "");
+}
+
+void GraphEngine::destroyMeter(uint32_t node_id) {
+    std::lock_guard<std::mutex> lk(m_meter_mutex);
+    auto it = m_meters.find(node_id);
+    if (it == m_meters.end()) return;
+    delete it->second;
+    m_meters.erase(it);
+}
+
+void GraphEngine::clearMeters() {
+    std::lock_guard<std::mutex> lk(m_meter_mutex);
+    for (auto &[id, ms] : m_meters)
+        delete ms;
+    m_meters.clear();
+}
+
+std::unordered_map<uint32_t, float> GraphEngine::getPeaks() const {
+    std::lock_guard<std::mutex> lk(m_meter_mutex);
+    std::unordered_map<uint32_t, float> out;
+    out.reserve(m_meters.size());
+    for (auto &[id, ms] : m_meters)
+        out[id] = ms->peak.load(std::memory_order_relaxed);
+    return out;
+}
+
 // ============================================================================
 // Pending/sync helpers (ported from qpwgraph_pipewire.cpp)
 // ============================================================================
@@ -163,6 +295,11 @@ static void on_node_info(void *data, const struct pw_node_info *info) {
         }
     }
 
+    // On first ready: spin up a meter stream for this node
+    if (!nobj->node.ready && obj->engine_ref)
+        obj->engine_ref->createMeterIfNeeded(
+            nobj->node.id, nobj->node.mode, nobj->node.node_type, nobj->node.name);
+
     nobj->node.changed = true;
     nobj->node.ready = true;
 
@@ -435,6 +572,10 @@ static void on_registry_global(void *data,
     auto *engine = static_cast<GraphEngine*>(data);
 
     if (strcmp(type, PW_TYPE_INTERFACE_Node) == 0) {
+        // Skip our own meter streams (avoid graph clutter and infinite recursion)
+        const char *app = spa_dict_lookup(props, PW_KEY_APP_NAME);
+        if (app && strcmp(app, "pwweb-meter") == 0) return;
+
         // Parse node properties (ported from qpwgraph lines 444-489)
         const char *str = spa_dict_lookup(props, PW_KEY_NODE_DESCRIPTION);
         const char *nick = spa_dict_lookup(props, PW_KEY_NODE_NICK);
@@ -443,7 +584,6 @@ static void on_registry_global(void *data,
         if (!str || strlen(str) < 1) str = "node";
 
         std::string node_name;
-        const char *app = spa_dict_lookup(props, PW_KEY_APP_NAME);
         if (app && strlen(app) > 0 && strcmp(app, str) != 0) {
             node_name += app;
             node_name += '/';
@@ -708,6 +848,9 @@ void GraphEngine::close() {
     if (m_pw.loop)
         pw_thread_loop_stop(m_pw.loop);
 
+    // PW thread is stopped; safe to destroy meter streams now
+    clearMeters();
+
     if (m_pw.registry) {
         spa_hook_remove(&m_pw.registry_listener);
         pw_proxy_destroy((pw_proxy*)m_pw.registry);
@@ -796,13 +939,14 @@ void GraphEngine::removeObject(uint32_t id) {
         }
     }
 
-    // If it's a node, remove all its ports
+    // If it's a node, remove all its ports and tear down its meter
     if (obj->type == Object::ObjNode) {
         auto *nobj = static_cast<NodeObj*>(obj);
         auto port_ids_copy = nobj->node.port_ids;
         for (uint32_t pid : port_ids_copy) {
             removeObject(pid);
         }
+        destroyMeter(id);
     }
 
     // If it's a link, remove from output port's link list

src/graph_engine.h

@@ -16,6 +16,7 @@ struct pw_thread_loop;
 struct pw_context;
 struct pw_core;
 struct pw_registry;
+struct pw_stream;
 
 namespace pwgraph {
 
@@ -135,6 +136,35 @@ public:
     uint32_t midiPortType()  const { return m_midi_type; }
     uint32_t videoPortType() const { return m_video_type; }
     uint32_t otherPortType() const { return m_other_type; }
+
+    // -----------------------------------------------------------------------
+    // Level metering
+    // -----------------------------------------------------------------------
+    struct MeterStream {
+        uint32_t            node_id   = 0;
+        struct pw_stream   *stream    = nullptr;
+        struct spa_hook     listener  = {};
+        std::atomic<float>  peak      {0.0f};   // instantaneous linear peak
+        GraphEngine        *engine_ref = nullptr;
+
+        MeterStream() = default;
+        ~MeterStream();
+        MeterStream(const MeterStream&) = delete;
+        MeterStream& operator=(const MeterStream&) = delete;
+    };
+
+    // Called from on_node_info when a node first becomes ready
+    void createMeterIfNeeded(uint32_t node_id, PortMode mode, NodeType ntype,
+                             const std::string& name);
+    void destroyMeter(uint32_t node_id);
+    void clearMeters();
+
+    // Returns linear peak (0–1) per node_id; called from HTTP thread
+    std::unordered_map<uint32_t, float> getPeaks() const;
+
+private:
+    std::unordered_map<uint32_t, MeterStream*> m_meters;
+    mutable std::mutex m_meter_mutex;
 };
 
 } // namespace pwgraph

src/web_server.cpp

@@ -302,6 +302,21 @@ void WebServer::setupRoutes() {
         res.set_header("Access-Control-Allow-Origin", "*");
     });
 
+    // Level meters: GET /api/peaks → {"<node_id>": <linear_peak_0_1>, ...}
+    m_http.Get("/api/peaks", [this](const httplib::Request &, httplib::Response &res) {
+        auto peaks = m_engine.getPeaks();
+        std::string json = "{";
+        bool first = true;
+        for (auto &[id, peak] : peaks) {
+            if (!first) json += ",";
+            json += "\"" + std::to_string(id) + "\":" + std::to_string(peak);
+            first = false;
+        }
+        json += "}";
+        res.set_content(json, "application/json");
+        res.set_header("Access-Control-Allow-Origin", "*");
+    });
+
     // REST API: POST /api/connect
     m_http.Post("/api/connect", [this](const httplib::Request &req, httplib::Response &res) {
         uint32_t out_id = 0, in_id = 0;