headroom

public

Star0

Fork 0

Watch0

3 commits 1 branch 0 tags 0 releases521.6 KiB

Commit `dd6b8e786e1f`

tarball

atagen <boss@atagen.co>2026-06-02 05:09 UTC

8 files changed+1657-0

ipc: uds control protocol and blocking client

Added file: crates/headroom-client/Cargo.toml+20 additions@@ -0,0 +1,20 @@
+[package]
+name = "headroom-client"
+description = "automatic loudness and per-app volume control (pipewire)"
+version.workspace = true
+edition.workspace = true
+license = "MPL-2.0"
+homepage.workspace = true
+repository.workspace = true
+authors.workspace = true
+
+[dependencies]
+headroom-ipc = { workspace = true }
+serde = { workspace = true }
+serde_json = { workspace = true }
+thiserror = { workspace = true }
+
+[features]
+default = []
+# reserved for an eventual tokio-based async client
+async = []

Added file: crates/headroom-client/src/client.rs+546 additions@@ -0,0 +1,546 @@
+//! blocking [`Client`].
+
+use std::collections::VecDeque;
+use std::io::{BufReader, BufWriter};
+use std::os::unix::net::UnixStream;
+use std::path::{Path, PathBuf};
+
+use serde::de::DeserializeOwned;
+
+use headroom_ipc::{
+    default_socket_path, Codec, Event, HelloData, Op, ProtoError, Request, Response,
+    ResponsePayload, Route, ServerFrame, Status, Topic,
+};
+
+/// errors from the blocking client.
+#[derive(Debug, thiserror::Error)]
+pub enum ClientError {
+    #[error("ipc: {0}")]
+    Ipc(#[from] headroom_ipc::Error),
+
+    /// server's first frame was not the `hello` event.
+    #[error("expected hello event from server, got {0}")]
+    BadHello(String),
+
+    /// response id we never issued.
+    #[error("response with unknown id {0}")]
+    UnknownResponseId(u64),
+
+    /// protocol-level error for an op.
+    #[error("server error: {0}")]
+    Protocol(#[from] ProtoError),
+
+    #[error("no default socket path (XDG_RUNTIME_DIR unset and /proc/self/status unreadable)")]
+    NoDefaultPath,
+
+    /// typed-helper response failed to deserialize into the expected shape.
+    #[error("response shape mismatch: {0}")]
+    DecodeResult(serde_json::Error),
+}
+
+/// blocking client for the headroom control protocol.
+///
+/// single-threaded by construction; for concurrent request/response and event
+/// consumption, open two connections.
+pub struct Client {
+    reader: BufReader<UnixStream>,
+    writer: BufWriter<UnixStream>,
+    codec: Codec,
+    next_id: u64,
+    pending_events: VecDeque<Event>,
+    hello: HelloData,
+    socket_path: PathBuf,
+}
+
+impl Client {
+    /// connect at the default socket path.
+    pub fn connect() -> Result<Self, ClientError> {
+        let path = default_socket_path().ok_or(ClientError::NoDefaultPath)?;
+        Self::connect_at(&path)
+    }
+
+    /// connect at the given socket path.
+    pub fn connect_at(path: &Path) -> Result<Self, ClientError> {
+        let stream = UnixStream::connect(path).map_err(|e| ClientError::Ipc(e.into()))?;
+        let reader_half = stream.try_clone().map_err(|e| ClientError::Ipc(e.into()))?;
+        let writer_half = stream;
+
+        let mut me = Self {
+            reader: BufReader::new(reader_half),
+            writer: BufWriter::new(writer_half),
+            codec: Codec::new(),
+            next_id: 1,
+            pending_events: VecDeque::new(),
+            // placeholder; populated by handshake below
+            hello: HelloData {
+                daemon: String::new(),
+                version: String::new(),
+                protocol: 0,
+            },
+            socket_path: path.to_path_buf(),
+        };
+
+        me.handshake()?;
+        Ok(me)
+    }
+
+    fn handshake(&mut self) -> Result<(), ClientError> {
+        let frame: ServerFrame = self.codec.read(&mut self.reader)?;
+        match frame {
+            ServerFrame::Event(ev)
+                if ev.topic == Topic::Control && ev.event.as_str() == "hello" =>
+            {
+                let hello: HelloData =
+                    serde_json::from_value(ev.data).map_err(ClientError::DecodeResult)?;
+                self.hello = hello;
+            }
+            ServerFrame::Event(ev) => {
+                return Err(ClientError::BadHello(format!(
+                    "{} event on {}",
+                    ev.event, ev.topic
+                )))
+            }
+            ServerFrame::Response(r) => {
+                return Err(ClientError::BadHello(format!("response id={}", r.id)))
+            }
+        }
+
+        // best effort for older daemons
+        self.send_raw(Op::Hello {
+            protocol: headroom_ipc::PROTOCOL_VERSION,
+        })?;
+        Ok(())
+    }
+
+    /// `hello` payload received on connect.
+    #[must_use]
+    pub fn hello(&self) -> &HelloData {
+        &self.hello
+    }
+
+    /// socket path this client is connected to.
+    #[must_use]
+    pub fn socket_path(&self) -> &Path {
+        &self.socket_path
+    }
+
+    fn alloc_id(&mut self) -> u64 {
+        let id = self.next_id;
+        // wrap: u64::MAX requests on one connection won't happen, but be correct
+        self.next_id = self.next_id.wrapping_add(1);
+        id
+    }
+
+    /// send a request, block until the paired response arrives.
+    ///
+    /// stray events received meanwhile are queued for [`next_event`](Self::next_event).
+    pub fn send(&mut self, op: Op) -> Result<serde_json::Value, ClientError> {
+        let payload = self.send_raw(op)?;
+        match payload {
+            ResponsePayload::Ok { result } => Ok(result),
+            ResponsePayload::Err { error } => Err(ClientError::Protocol(error)),
+        }
+    }
+
+    /// like [`send`](Self::send) but returns the raw [`ResponsePayload`] —
+    /// keeps a protocol-level error in-band instead of as [`ClientError::Protocol`].
+    pub fn send_raw(&mut self, op: Op) -> Result<ResponsePayload, ClientError> {
+        let id = self.alloc_id();
+        let req = Request::new(id, op);
+        self.codec.write(&mut self.writer, &req)?;
+
+        loop {
+            let frame: ServerFrame = self.codec.read(&mut self.reader)?;
+            match frame {
+                ServerFrame::Response(Response {
+                    id: rid,
+                    payload: _,
+                }) if rid != id => {
+                    return Err(ClientError::UnknownResponseId(rid));
+                }
+                ServerFrame::Response(Response { payload, .. }) => return Ok(payload),
+                ServerFrame::Event(ev) => {
+                    self.pending_events.push_back(ev);
+                }
+            }
+        }
+    }
+
+    /// block until the next event arrives. drains the queue before reading the socket.
+    ///
+    /// an unsolicited response (no matching request) is rejected as
+    /// [`ClientError::UnknownResponseId`].
+    pub fn next_event(&mut self) -> Result<Event, ClientError> {
+        if let Some(ev) = self.pending_events.pop_front() {
+            return Ok(ev);
+        }
+        match self.codec.read::<_, ServerFrame>(&mut self.reader)? {
+            ServerFrame::Event(ev) => Ok(ev),
+            ServerFrame::Response(r) => Err(ClientError::UnknownResponseId(r.id)),
+        }
+    }
+
+    /// pop a queued event without blocking. does **not** read from the socket.
+    pub fn pending_event(&mut self) -> Option<Event> {
+        self.pending_events.pop_front()
+    }
+
+    // typed convenience wrappers
+
+    fn send_into<T: DeserializeOwned>(&mut self, op: Op) -> Result<T, ClientError> {
+        let value = self.send(op)?;
+        serde_json::from_value(value).map_err(ClientError::DecodeResult)
+    }
+
+    /// `status`
+    pub fn status(&mut self) -> Result<Status, ClientError> {
+        self.send_into(Op::Status)
+    }
+
+    /// `profile.list`
+    pub fn profile_list(&mut self) -> Result<Vec<headroom_ipc::ProfileInfo>, ClientError> {
+        #[derive(serde::Deserialize)]
+        struct Body {
+            profiles: Vec<headroom_ipc::ProfileInfo>,
+        }
+        let body: Body = self.send_into(Op::ProfileList)?;
+        Ok(body.profiles)
+    }
+
+    /// `profile.use`
+    pub fn profile_use(&mut self, name: &str) -> Result<String, ClientError> {
+        #[derive(serde::Deserialize)]
+        struct Body {
+            name: String,
+        }
+        let body: Body = self.send_into(Op::ProfileUse {
+            name: name.to_owned(),
+        })?;
+        Ok(body.name)
+    }
+
+    /// `profile.show`
+    pub fn profile_show(
+        &mut self,
+        name: Option<&str>,
+    ) -> Result<serde_json::Value, ClientError> {
+        self.send(Op::ProfileShow {
+            name: name.map(String::from),
+        })
+    }
+
+    /// `profile.reload`
+    pub fn profile_reload(&mut self) -> Result<Vec<String>, ClientError> {
+        #[derive(serde::Deserialize)]
+        struct Body {
+            reloaded: Vec<String>,
+        }
+        let body: Body = self.send_into(Op::ProfileReload)?;
+        Ok(body.reloaded)
+    }
+
+    /// `route.list`
+    pub fn route_list(&mut self) -> Result<headroom_ipc::RouteList, ClientError> {
+        self.send_into(Op::RouteList)
+    }
+
+    /// `route.set`
+    pub fn route_set(&mut self, app: &str, to: Route) -> Result<(), ClientError> {
+        let _: serde_json::Value = self.send(Op::RouteSet {
+            app: app.to_owned(),
+            to,
+        })?;
+        Ok(())
+    }
+
+    /// `route.unset`
+    pub fn route_unset(&mut self, app: &str) -> Result<(), ClientError> {
+        let _: serde_json::Value = self.send(Op::RouteUnset {
+            app: app.to_owned(),
+        })?;
+        Ok(())
+    }
+
+    /// `route.stream`
+    pub fn route_stream(&mut self, node_id: u32, to: Route) -> Result<(), ClientError> {
+        let _: serde_json::Value = self.send(Op::RouteStream { node_id, to })?;
+        Ok(())
+    }
+
+    /// `setting.get`
+    pub fn setting_get(&mut self, key: &str) -> Result<serde_json::Value, ClientError> {
+        #[derive(serde::Deserialize)]
+        struct Body {
+            #[allow(dead_code)]
+            key: String,
+            value: serde_json::Value,
+        }
+        let body: Body = self.send_into(Op::SettingGet {
+            key: key.to_owned(),
+        })?;
+        Ok(body.value)
+    }
+
+    /// `setting.set`
+    pub fn setting_set(
+        &mut self,
+        key: &str,
+        value: serde_json::Value,
+    ) -> Result<(), ClientError> {
+        let _: serde_json::Value = self.send(Op::SettingSet {
+            key: key.to_owned(),
+            value,
+        })?;
+        Ok(())
+    }
+
+    /// `setting.clear` — returns whether an override was present.
+    pub fn setting_clear(&mut self, key: &str) -> Result<bool, ClientError> {
+        #[derive(serde::Deserialize)]
+        struct Body {
+            cleared: bool,
+        }
+        let body: Body = self.send_into(Op::SettingClear {
+            key: key.to_owned(),
+        })?;
+        Ok(body.cleared)
+    }
+
+    /// `setting.reset` — returns how many overrides were cleared.
+    pub fn setting_reset(&mut self) -> Result<usize, ClientError> {
+        #[derive(serde::Deserialize)]
+        struct Body {
+            cleared: usize,
+        }
+        let body: Body = self.send_into(Op::SettingReset)?;
+        Ok(body.cleared)
+    }
+
+    /// `bypass.set`
+    pub fn bypass_set(&mut self, enabled: bool) -> Result<(), ClientError> {
+        let _: serde_json::Value = self.send(Op::BypassSet { enabled })?;
+        Ok(())
+    }
+
+    /// `per-app.list`
+    pub fn layer_a_list(
+        &mut self,
+    ) -> Result<Vec<headroom_ipc::LayerASnapshot>, ClientError> {
+        #[derive(serde::Deserialize)]
+        struct Body {
+            layer_a: Vec<headroom_ipc::LayerASnapshot>,
+        }
+        let body: Body = self.send_into(Op::LayerAList)?;
+        Ok(body.layer_a)
+    }
+
+    /// `per-app.set`
+    pub fn per_app_set(&mut self, app: &str, enabled: bool) -> Result<(), ClientError> {
+        let _: serde_json::Value = self.send(Op::PerAppSet {
+            app: app.to_owned(),
+            enabled,
+        })?;
+        Ok(())
+    }
+
+    /// `per-app.master`
+    pub fn per_app_master(&mut self, enabled: bool) -> Result<(), ClientError> {
+        let _: serde_json::Value = self.send(Op::PerAppMaster { enabled })?;
+        Ok(())
+    }
+
+    /// `per-app.reset`
+    pub fn layer_a_reset(&mut self, node_id: u32) -> Result<(), ClientError> {
+        let _: serde_json::Value = self.send(Op::LayerAReset { node_id })?;
+        Ok(())
+    }
+
+    /// `subscribe`
+    pub fn subscribe(&mut self, topics: &[Topic]) -> Result<Vec<Topic>, ClientError> {
+        #[derive(serde::Deserialize)]
+        struct Body {
+            subscribed: Vec<Topic>,
+        }
+        let body: Body = self.send_into(Op::Subscribe {
+            topics: topics.to_vec(),
+        })?;
+        Ok(body.subscribed)
+    }
+
+    /// `unsubscribe`
+    pub fn unsubscribe(&mut self, topics: &[Topic]) -> Result<Vec<Topic>, ClientError> {
+        #[derive(serde::Deserialize)]
+        struct Body {
+            unsubscribed: Vec<Topic>,
+        }
+        let body: Body = self.send_into(Op::Unsubscribe {
+            topics: topics.to_vec(),
+        })?;
+        Ok(body.unsubscribed)
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use std::io::{BufReader, BufWriter};
+    use std::os::unix::net::UnixStream;
+    use std::thread;
+
+    use headroom_ipc::{Codec, Event, HelloData, Op, Request, Response, ServerFrame, Topic};
+
+    /// A tiny in-process server that runs on the other end of a
+    /// `UnixStream::pair`. Knows just enough to exercise the client.
+    fn spawn_test_server() -> (UnixStream, thread::JoinHandle<()>) {
+        let (a, b) = UnixStream::pair().unwrap();
+        let server_handle = thread::spawn(move || {
+            let codec = Codec::new();
+            let read_side = b.try_clone().unwrap();
+            let mut reader = BufReader::new(read_side);
+            let mut writer = BufWriter::new(b);
+
+            // Send hello.
+            let hello = Event::new(
+                Topic::Control,
+                "hello",
+                &HelloData {
+                    daemon: "headroom".into(),
+                    version: "0.1.0-test".into(),
+                    protocol: headroom_ipc::PROTOCOL_VERSION,
+                },
+            )
+            .unwrap();
+            codec
+                .write(&mut writer, &ServerFrame::Event(hello))
+                .unwrap();
+
+            // Serve one round.
+            loop {
+                let req: Request = match codec.read(&mut reader) {
+                    Ok(r) => r,
+                    Err(_) => return,
+                };
+                let resp = match req.op {
+                    Op::Status => Response::ok(
+                        req.id,
+                        &serde_json::json!({
+                            "version": "0.1.0-test",
+                            "protocol": headroom_ipc::PROTOCOL_VERSION,
+                            "uptime_s": 0u64,
+                            "profile": "default",
+                            "bypass": false,
+                            "sinks": {
+                                "processed": {"ready": false},
+                                "real":      {"ready": false},
+                            },
+                            "streams": []
+                        }),
+                    )
+                    .unwrap(),
+                    Op::ProfileUse { name } => {
+                        Response::ok(req.id, &serde_json::json!({ "name": name })).unwrap()
+                    }
+                    Op::Subscribe { topics } => {
+                        // Acknowledge, then push one event of each
+                        // subscribed topic so the client can demonstrate
+                        // event handling.
+                        let body = serde_json::json!({ "subscribed": &topics });
+                        let resp = Response::ok(req.id, &body).unwrap();
+                        codec
+                            .write(&mut writer, &ServerFrame::Response(resp.clone()))
+                            .unwrap();
+
+                        for t in topics {
+                            let ev = Event::new(t, "tick", &serde_json::json!({})).unwrap();
+                            codec.write(&mut writer, &ServerFrame::Event(ev)).unwrap();
+                        }
+                        continue;
+                    }
+                    _ => Response::ok(req.id, &serde_json::Value::Null).unwrap(),
+                };
+                codec
+                    .write(&mut writer, &ServerFrame::Response(resp))
+                    .unwrap();
+            }
+        });
+        (a, server_handle)
+    }
+
+    fn client_on(stream: UnixStream) -> Client {
+        let reader_half = stream.try_clone().unwrap();
+        let writer_half = stream;
+        let mut me = Client {
+            reader: BufReader::new(reader_half),
+            writer: BufWriter::new(writer_half),
+            codec: Codec::new(),
+            next_id: 1,
+            pending_events: VecDeque::new(),
+            hello: HelloData {
+                daemon: String::new(),
+                version: String::new(),
+                protocol: 0,
+            },
+            socket_path: PathBuf::from("<test>"),
+        };
+        me.handshake().unwrap();
+        me
+    }
+
+    #[test]
+    fn handshake_then_status() {
+        let (client_sock, _server) = spawn_test_server();
+        let mut client = client_on(client_sock);
+        assert_eq!(client.hello().daemon, "headroom");
+        assert_eq!(client.hello().protocol, headroom_ipc::PROTOCOL_VERSION);
+
+        let status = client.status().unwrap();
+        assert_eq!(status.profile, "default");
+        assert!(!status.bypass);
+    }
+
+    #[test]
+    fn profile_use_returns_name() {
+        let (client_sock, _server) = spawn_test_server();
+        let mut client = client_on(client_sock);
+        let name = client.profile_use("night").unwrap();
+        assert_eq!(name, "night");
+    }
+
+    #[test]
+    fn subscribe_then_consume_event() {
+        let (client_sock, _server) = spawn_test_server();
+        let mut client = client_on(client_sock);
+        let acked = client.subscribe(&[Topic::Meters]).unwrap();
+        assert_eq!(acked, vec![Topic::Meters]);
+
+        let ev = client.next_event().unwrap();
+        assert_eq!(ev.topic, Topic::Meters);
+        assert_eq!(ev.event, "tick");
+    }
+
+    #[test]
+    fn events_interleaved_during_request_are_queued() {
+        // The test server pushes events *after* the subscribe response,
+        // so let's check that requesting another op afterwards drains
+        // them through the queue.
+        let (client_sock, _server) = spawn_test_server();
+        let mut client = client_on(client_sock);
+        client.subscribe(&[Topic::Meters, Topic::Profile]).unwrap();
+
+        // Now issue another request. The server hasn't sent the events
+        // until we read more, but our client will keep reading.
+        let status = client.status().unwrap();
+        assert_eq!(status.profile, "default");
+
+        // We may have buffered events from the prior subscribe and from
+        // any in flight; drain them.
+        let mut topics = Vec::new();
+        while let Some(ev) = client.pending_event() {
+            topics.push(ev.topic);
+        }
+        // The events arrived between the subscribe-ack and the status
+        // response; both should be queued.
+        assert!(topics.contains(&Topic::Meters));
+        assert!(topics.contains(&Topic::Profile));
+    }
+}

Added file: crates/headroom-client/src/lib.rs+15 additions@@ -0,0 +1,15 @@
+//! blocking client for the headroom control protocol. entry point: [`Client`].
+
+#![forbid(unsafe_code)]
+#![warn(missing_docs)]
+
+mod client;
+
+pub use client::{Client, ClientError};
+
+pub use headroom_ipc::{
+    default_socket_path, Codec, DaemonEvent, Error as IpcError, ErrorCode, Event, HelloData,
+    LayerALevel, LayerASnapshot, MeterTick, Op, ProfileEvent, ProfileInfo, ProtoError, Request,
+    Response, ResponsePayload, Route, RouteList, RouteRule, RouteRuleMatch, RoutingEvent,
+    ServerFrame, SinkInfo, Sinks, Status, StreamRoute, Topic, PROTOCOL_VERSION,
+};

Added file: crates/headroom-ipc/Cargo.toml+14 additions@@ -0,0 +1,14 @@
+[package]
+name = "headroom-ipc"
+description = "automatic loudness and per-app volume control (pipewire)"
+version.workspace = true
+edition.workspace = true
+license = "MPL-2.0"
+homepage.workspace = true
+repository.workspace = true
+authors.workspace = true
+
+[dependencies]
+serde = { workspace = true }
+serde_json = { workspace = true }
+thiserror = { workspace = true }

Added file: crates/headroom-ipc/src/codec.rs+248 additions@@ -0,0 +1,248 @@
+//! length-prefixed json framing: 4-byte big-endian length, then that many bytes of utf-8 json.
+
+use std::io::{Read, Write};
+
+use serde::{de::DeserializeOwned, Serialize};
+
+use crate::error::Error;
+
+/// default frame payload cap.
+pub const DEFAULT_MAX_FRAME_BYTES: usize = 1024 * 1024; // 1 MiB
+
+/// floor on the frame cap; a smaller cap is misuse, not normal traffic.
+pub const MIN_MAX_FRAME_BYTES: usize = 256;
+
+/// stateless framing codec; owns no buffers, callers supply reader/writer.
+#[derive(Debug, Clone, Copy)]
+pub struct Codec {
+    max_frame_bytes: usize,
+}
+
+impl Default for Codec {
+    fn default() -> Self {
+        Self {
+            max_frame_bytes: DEFAULT_MAX_FRAME_BYTES,
+        }
+    }
+}
+
+impl Codec {
+    #[must_use]
+    pub fn new() -> Self {
+        Self::default()
+    }
+
+    /// cap clamped up to [`MIN_MAX_FRAME_BYTES`].
+    #[must_use]
+    pub fn with_max_frame_size(bytes: usize) -> Self {
+        Self {
+            max_frame_bytes: bytes.max(MIN_MAX_FRAME_BYTES),
+        }
+    }
+
+    #[must_use]
+    pub fn max_frame_bytes(self) -> usize {
+        self.max_frame_bytes
+    }
+
+    /// # Errors
+    /// - [`Error::Json`] on serialize failure.
+    /// - [`Error::FrameTooLarge`] if the serialized form exceeds the cap.
+    /// - [`Error::Io`] on write failure.
+    pub fn write<W: Write, T: Serialize>(self, mut w: W, msg: &T) -> Result<(), Error> {
+        // serialize first to know the exact length
+        let buf = serde_json::to_vec(msg)?;
+        if buf.len() > self.max_frame_bytes {
+            return Err(Error::FrameTooLarge {
+                actual: buf.len(),
+                limit: self.max_frame_bytes,
+            });
+        }
+        let len = u32::try_from(buf.len()).expect("buf.len() <= max_frame_bytes <= u32::MAX");
+        w.write_all(&len.to_be_bytes())?;
+        w.write_all(&buf)?;
+        w.flush()?;
+        Ok(())
+    }
+
+    /// # Errors
+    /// - [`Error::Closed`] on EOF before any length-prefix byte (graceful close).
+    /// - [`Error::Io`] on partial reads or other i/o failure.
+    /// - [`Error::FrameTooLarge`] if the announced length exceeds the cap.
+    /// - [`Error::Json`] on deserialize failure.
+    pub fn read<R: Read, T: DeserializeOwned>(self, mut r: R) -> Result<T, Error> {
+        let mut len_buf = [0u8; 4];
+        match read_full_or_eof(&mut r, &mut len_buf)? {
+            ReadOutcome::Full => {}
+            ReadOutcome::ZeroAtStart => return Err(Error::Closed),
+        }
+        let len = u32::from_be_bytes(len_buf) as usize;
+        if len > self.max_frame_bytes {
+            return Err(Error::FrameTooLarge {
+                actual: len,
+                limit: self.max_frame_bytes,
+            });
+        }
+        let mut buf = vec![0u8; len];
+        r.read_exact(&mut buf)?;
+        Ok(serde_json::from_slice(&buf)?)
+    }
+}
+
+enum ReadOutcome {
+    Full,
+    ZeroAtStart,
+}
+
+fn read_full_or_eof<R: Read>(r: &mut R, buf: &mut [u8]) -> Result<ReadOutcome, std::io::Error> {
+    let mut read = 0;
+    while read < buf.len() {
+        match r.read(&mut buf[read..]) {
+            Ok(0) => {
+                if read == 0 {
+                    return Ok(ReadOutcome::ZeroAtStart);
+                }
+                return Err(std::io::Error::new(
+                    std::io::ErrorKind::UnexpectedEof,
+                    "eof mid-length-prefix",
+                ));
+            }
+            Ok(n) => read += n,
+            Err(e) if e.kind() == std::io::ErrorKind::Interrupted => continue,
+            Err(e) => return Err(e),
+        }
+    }
+    Ok(ReadOutcome::Full)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::{Event, Op, Request, Response, ServerFrame, Topic};
+    use std::io::Cursor;
+
+    #[test]
+    fn write_read_request_roundtrip() {
+        let codec = Codec::new();
+        let req = Request::new(
+            42,
+            Op::ProfileUse {
+                name: "night".into(),
+            },
+        );
+
+        let mut buf = Vec::new();
+        codec.write(&mut buf, &req).unwrap();
+
+        let mut cur = Cursor::new(&buf);
+        let back: Request = codec.read(&mut cur).unwrap();
+        assert_eq!(back, req);
+    }
+
+    #[test]
+    fn server_frame_round_trip_response() {
+        let codec = Codec::new();
+        let resp = Response::ok(1, &serde_json::json!({"name": "default"})).unwrap();
+        let frame = ServerFrame::Response(resp.clone());
+
+        let mut buf = Vec::new();
+        codec.write(&mut buf, &frame).unwrap();
+        let mut cur = Cursor::new(&buf);
+        let back: ServerFrame = codec.read(&mut cur).unwrap();
+        match back {
+            ServerFrame::Response(r) => assert_eq!(r, resp),
+            ServerFrame::Event(_) => panic!("decoded as event"),
+        }
+    }
+
+    #[test]
+    fn server_frame_round_trip_event() {
+        let codec = Codec::new();
+        let ev = Event::new(Topic::Daemon, "shutdown", &serde_json::json!({})).unwrap();
+        let frame = ServerFrame::Event(ev.clone());
+
+        let mut buf = Vec::new();
+        codec.write(&mut buf, &frame).unwrap();
+        let mut cur = Cursor::new(&buf);
+        let back: ServerFrame = codec.read(&mut cur).unwrap();
+        match back {
+            ServerFrame::Event(e) => assert_eq!(e, ev),
+            ServerFrame::Response(_) => panic!("decoded as response"),
+        }
+    }
+
+    #[test]
+    fn rejects_oversized_frames_on_write() {
+        let codec = Codec::with_max_frame_size(MIN_MAX_FRAME_BYTES);
+        // A big string that will serialize > 256 bytes.
+        let req = Request::new(
+            1,
+            Op::SettingSet {
+                key: "x".into(),
+                value: serde_json::Value::String("a".repeat(1024)),
+            },
+        );
+        let mut buf = Vec::new();
+        let err = codec.write(&mut buf, &req).unwrap_err();
+        assert!(matches!(err, Error::FrameTooLarge { .. }));
+    }
+
+    #[test]
+    fn rejects_oversized_frames_on_read() {
+        let codec = Codec::with_max_frame_size(MIN_MAX_FRAME_BYTES);
+        // Hand-craft a length prefix that exceeds the cap.
+        let mut buf = Vec::new();
+        let bad_len: u32 = MIN_MAX_FRAME_BYTES as u32 + 1;
+        buf.extend_from_slice(&bad_len.to_be_bytes());
+        // No need to follow with payload; we expect early rejection.
+        let mut cur = Cursor::new(&buf);
+        let err = codec.read::<_, serde_json::Value>(&mut cur).unwrap_err();
+        assert!(matches!(err, Error::FrameTooLarge { .. }));
+    }
+
+    #[test]
+    fn graceful_eof_at_frame_boundary() {
+        let codec = Codec::new();
+        let mut cur = Cursor::new(Vec::<u8>::new());
+        let err = codec.read::<_, Request>(&mut cur).unwrap_err();
+        assert!(matches!(err, Error::Closed));
+    }
+
+    #[test]
+    fn mid_frame_eof_is_io_error() {
+        let codec = Codec::new();
+        // Half a length prefix.
+        let mut cur = Cursor::new(vec![0u8, 0u8]);
+        let err = codec.read::<_, Request>(&mut cur).unwrap_err();
+        assert!(matches!(err, Error::Io(_)));
+    }
+
+    #[test]
+    fn rejects_invalid_json_payload() {
+        let codec = Codec::new();
+        let payload = b"not-json";
+        let mut buf = Vec::new();
+        buf.extend_from_slice(&(payload.len() as u32).to_be_bytes());
+        buf.extend_from_slice(payload);
+        let mut cur = Cursor::new(&buf);
+        let err = codec.read::<_, Request>(&mut cur).unwrap_err();
+        assert!(matches!(err, Error::Json(_)));
+    }
+
+    #[test]
+    fn back_to_back_frames() {
+        let codec = Codec::new();
+        let a = Request::new(1, Op::Status);
+        let b = Request::new(2, Op::ProfileList);
+
+        let mut buf = Vec::new();
+        codec.write(&mut buf, &a).unwrap();
+        codec.write(&mut buf, &b).unwrap();
+
+        let mut cur = Cursor::new(&buf);
+        let ra: Request = codec.read(&mut cur).unwrap();
+        let rb: Request = codec.read(&mut cur).unwrap();
+        assert_eq!(ra, a);
+        assert_eq!(rb, b);
+    }
+}

Added file: crates/headroom-ipc/src/error.rs+108 additions@@ -0,0 +1,108 @@
+//! error types.
+
+use serde::{Deserialize, Serialize};
+
+/// wire error code in `error.code`. adding variants is non-breaking; removing/renaming bumps `PROTOCOL_VERSION`.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
+#[serde(rename_all = "SCREAMING_SNAKE_CASE")]
+#[non_exhaustive]
+pub enum ErrorCode {
+    /// malformed framing or non-json payload; connection is closed.
+    InvalidFrame,
+    /// valid json, no known message shape.
+    InvalidMessage,
+    /// `op` names no known operation.
+    UnknownOp,
+    /// `args` missing a field, wrong type, or out of range.
+    InvalidArgs,
+    /// named profile / app / stream / setting key does not exist.
+    NotFound,
+    /// would violate an invariant.
+    Conflict,
+    /// daemon transiently cannot serve the request.
+    Busy,
+    /// server-side bug; `message` carries detail.
+    Internal,
+}
+
+impl ErrorCode {
+    #[must_use]
+    pub fn as_str(self) -> &'static str {
+        match self {
+            ErrorCode::InvalidFrame => "INVALID_FRAME",
+            ErrorCode::InvalidMessage => "INVALID_MESSAGE",
+            ErrorCode::UnknownOp => "UNKNOWN_OP",
+            ErrorCode::InvalidArgs => "INVALID_ARGS",
+            ErrorCode::NotFound => "NOT_FOUND",
+            ErrorCode::Conflict => "CONFLICT",
+            ErrorCode::Busy => "BUSY",
+            ErrorCode::Internal => "INTERNAL",
+        }
+    }
+}
+
+impl std::fmt::Display for ErrorCode {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.write_str(self.as_str())
+    }
+}
+
+/// error payload inside a `Response`.
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+pub struct ProtoError {
+    pub code: ErrorCode,
+    /// not stable; do not pattern match.
+    pub message: String,
+}
+
+impl ProtoError {
+    #[must_use]
+    pub fn new(code: ErrorCode, message: impl Into<String>) -> Self {
+        Self {
+            code,
+            message: message.into(),
+        }
+    }
+}
+
+impl std::fmt::Display for ProtoError {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        write!(f, "{}: {}", self.code, self.message)
+    }
+}
+
+impl std::error::Error for ProtoError {}
+
+/// errors from the codec and helpers.
+#[derive(Debug, thiserror::Error)]
+pub enum Error {
+    #[error("io: {0}")]
+    Io(#[from] std::io::Error),
+
+    #[error("json: {0}")]
+    Json(#[from] serde_json::Error),
+
+    #[error("frame too large: {actual} bytes (limit {limit})")]
+    FrameTooLarge { actual: usize, limit: usize },
+
+    #[error("protocol: {0}")]
+    Protocol(#[from] ProtoError),
+
+    /// e.g. a response with a mismatched id.
+    #[error("unexpected frame: {0}")]
+    UnexpectedFrame(String),
+
+    #[error("connection closed")]
+    Closed,
+}
+
+impl Error {
+    /// true when the connection should be torn down.
+    #[must_use]
+    pub fn is_fatal(&self) -> bool {
+        matches!(
+            self,
+            Error::Io(_) | Error::FrameTooLarge { .. } | Error::Closed
+        )
+    }
+}

Added file: crates/headroom-ipc/src/lib.rs+57 additions@@ -0,0 +1,57 @@
+//! headroom control-protocol types and framing. spec: `IPC.md` (repo root); this is its rust binding.
+
+#![forbid(unsafe_code)]
+#![warn(missing_docs)]
+
+mod codec;
+mod error;
+mod proto;
+
+pub use codec::{Codec, DEFAULT_MAX_FRAME_BYTES, MIN_MAX_FRAME_BYTES};
+pub use error::{Error, ErrorCode, ProtoError};
+pub use proto::{
+    DaemonEvent, Event, HelloData, LayerALevel, LayerASnapshot, MeterTick, Op, ProfileEvent,
+    ProfileInfo, Request, Response, ResponsePayload, Route, RouteList, RouteRule, RouteRuleMatch,
+    RoutingEvent, ServerFrame, SinkInfo, Sinks, Status, StreamRoute, Topic,
+};
+
+/// wire version; bumped only on incompatible changes.
+pub const PROTOCOL_VERSION: u32 = 1;
+
+/// socket dir stem; full path `${XDG_RUNTIME_DIR}/headroom/control.sock`, falling back to `/run/user/$UID/...`.
+pub const DEFAULT_SOCKET_DIR: &str = "headroom";
+
+pub const DEFAULT_SOCKET_NAME: &str = "control.sock";
+
+/// honours `XDG_RUNTIME_DIR`, else falls back to `/run/user/$UID/...`. `None` when neither is determinable.
+#[must_use]
+pub fn default_socket_path() -> Option<std::path::PathBuf> {
+    use std::path::PathBuf;
+    if let Ok(dir) = std::env::var("XDG_RUNTIME_DIR") {
+        if !dir.is_empty() {
+            return Some(
+                PathBuf::from(dir)
+                    .join(DEFAULT_SOCKET_DIR)
+                    .join(DEFAULT_SOCKET_NAME),
+            );
+        }
+    }
+    // uid from /proc/self/status: nix-free, dependency-light
+    let uid = read_self_uid()?;
+    Some(
+        PathBuf::from(format!("/run/user/{uid}"))
+            .join(DEFAULT_SOCKET_DIR)
+            .join(DEFAULT_SOCKET_NAME),
+    )
+}
+
+fn read_self_uid() -> Option<u32> {
+    let s = std::fs::read_to_string("/proc/self/status").ok()?;
+    for line in s.lines() {
+        if let Some(rest) = line.strip_prefix("Uid:") {
+            let first = rest.split_whitespace().next()?;
+            return first.parse().ok();
+        }
+    }
+    None
+}

Added file: crates/headroom-ipc/src/proto.rs+649 additions@@ -0,0 +1,649 @@
+//! protocol message types. spec: `IPC.md`.
+
+use std::collections::BTreeMap;
+
+use serde::{Deserialize, Serialize};
+use serde_json::Value;
+
+use crate::error::ProtoError;
+
+/// subscription topic.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
+#[serde(rename_all = "kebab-case")]
+#[non_exhaustive]
+pub enum Topic {
+    /// loudness / peak / GR telemetry.
+    Meters,
+    Profile,
+    Routing,
+    Daemon,
+    /// synthetic, for `hello`; clients never subscribe.
+    Control,
+}
+
+impl Topic {
+    #[must_use]
+    pub fn as_str(self) -> &'static str {
+        match self {
+            Topic::Meters => "meters",
+            Topic::Profile => "profile",
+            Topic::Routing => "routing",
+            Topic::Daemon => "daemon",
+            Topic::Control => "control",
+        }
+    }
+
+    /// topics clients may subscribe to (excludes `Control`).
+    #[must_use]
+    pub const fn subscribable() -> &'static [Topic] {
+        &[Topic::Meters, Topic::Profile, Topic::Routing, Topic::Daemon]
+    }
+}
+
+impl std::fmt::Display for Topic {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.write_str(self.as_str())
+    }
+}
+
+/// routing decision for one app or stream.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
+#[serde(rename_all = "lowercase")]
+pub enum Route {
+    /// through the processed (filtered) sink.
+    Processed,
+    /// straight to the real hardware sink; no processing, no extra graph hop.
+    Bypass,
+}
+
+impl Route {
+    #[must_use]
+    pub fn as_str(self) -> &'static str {
+        match self {
+            Route::Processed => "processed",
+            Route::Bypass => "bypass",
+        }
+    }
+}
+
+impl std::fmt::Display for Route {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.write_str(self.as_str())
+    }
+}
+
+/// all operations. wire form `{ "op": "<name>", "args": <args> }`; arg-less ops omit `args`.
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+#[serde(tag = "op", content = "args")]
+#[non_exhaustive]
+pub enum Op {
+    #[serde(rename = "status")]
+    Status,
+
+    #[serde(rename = "profile.list")]
+    ProfileList,
+
+    #[serde(rename = "profile.use")]
+    ProfileUse { name: String },
+
+    /// `name` omitted ⇒ active profile.
+    #[serde(rename = "profile.show")]
+    ProfileShow {
+        #[serde(default, skip_serializing_if = "Option::is_none")]
+        name: Option<String>,
+    },
+
+    /// reload all profile files from disk.
+    #[serde(rename = "profile.reload")]
+    ProfileReload,
+
+    #[serde(rename = "route.list")]
+    RouteList,
+
+    /// add/replace a persistent routing rule.
+    #[serde(rename = "route.set")]
+    RouteSet {
+        /// typically `application.process.binary`.
+        app: String,
+        to: Route,
+    },
+
+    #[serde(rename = "route.unset")]
+    RouteUnset { app: String },
+
+    /// one-shot reroute of a live stream.
+    #[serde(rename = "route.stream")]
+    RouteStream { node_id: u32, to: Route },
+
+    #[serde(rename = "setting.get")]
+    SettingGet {
+        /// dotted key, e.g. `compressor.threshold_db`.
+        key: String,
+    },
+
+    #[serde(rename = "setting.set")]
+    SettingSet { key: String, value: Value },
+
+    #[serde(rename = "setting.list")]
+    SettingList,
+
+    /// revert one key to the active profile's value.
+    #[serde(rename = "setting.clear")]
+    SettingClear { key: String },
+
+    /// revert all settings to the active profile; leaves route/per-app/bypass overrides intact.
+    #[serde(rename = "setting.reset")]
+    SettingReset,
+
+    /// global bypass kill switch.
+    #[serde(rename = "bypass.set")]
+    BypassSet { enabled: bool },
+
+    /// per-app (Layer A) controller state for managed streams.
+    #[serde(rename = "per-app.list")]
+    LayerAList,
+
+    /// Layer A enable for one app (persistent overlay override).
+    #[serde(rename = "per-app.set")]
+    PerAppSet {
+        /// process_binary or application_name.
+        app: String,
+        enabled: bool,
+    },
+
+    /// Layer A master switch (persistent overlay override).
+    #[serde(rename = "per-app.master")]
+    PerAppMaster { enabled: bool },
+
+    /// clear a stream's deference lock (user-ceiling / strict-mode) so the controller resumes.
+    #[serde(rename = "per-app.reset")]
+    LayerAReset { node_id: u32 },
+
+    #[serde(rename = "subscribe")]
+    Subscribe { topics: Vec<Topic> },
+
+    #[serde(rename = "unsubscribe")]
+    Unsubscribe { topics: Vec<Topic> },
+
+    /// optional client protocol handshake
+    #[serde(rename = "hello")]
+    Hello { protocol: u32 },
+}
+
+impl Op {
+    /// canonical wire name.
+    #[must_use]
+    pub fn name(&self) -> &'static str {
+        match self {
+            Op::Status => "status",
+            Op::ProfileList => "profile.list",
+            Op::ProfileUse { .. } => "profile.use",
+            Op::ProfileShow { .. } => "profile.show",
+            Op::ProfileReload => "profile.reload",
+            Op::RouteList => "route.list",
+            Op::RouteSet { .. } => "route.set",
+            Op::RouteUnset { .. } => "route.unset",
+            Op::RouteStream { .. } => "route.stream",
+            Op::SettingGet { .. } => "setting.get",
+            Op::SettingSet { .. } => "setting.set",
+            Op::SettingList => "setting.list",
+            Op::SettingClear { .. } => "setting.clear",
+            Op::SettingReset => "setting.reset",
+            Op::BypassSet { .. } => "bypass.set",
+            Op::LayerAList => "per-app.list",
+            Op::PerAppSet { .. } => "per-app.set",
+            Op::PerAppMaster { .. } => "per-app.master",
+            Op::LayerAReset { .. } => "per-app.reset",
+            Op::Subscribe { .. } => "subscribe",
+            Op::Unsubscribe { .. } => "unsubscribe",
+            Op::Hello { .. } => "hello",
+        }
+    }
+}
+
+/// client-to-server request.
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+pub struct Request {
+    /// client-chosen; echoed in the paired response.
+    pub id: u64,
+    /// flattened: contributes `op` and optional `args` alongside `id`.
+    #[serde(flatten)]
+    pub op: Op,
+}
+
+impl Request {
+    #[must_use]
+    pub fn new(id: u64, op: Op) -> Self {
+        Self { id, op }
+    }
+}
+
+/// server-to-client response, paired to a request by `id`.
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+pub struct Response {
+    pub id: u64,
+    #[serde(flatten)]
+    pub payload: ResponsePayload,
+}
+
+impl Response {
+    /// # Errors
+    /// serde error if `value` fails to serialize.
+    pub fn ok<T: Serialize>(id: u64, value: &T) -> Result<Self, serde_json::Error> {
+        Ok(Self {
+            id,
+            payload: ResponsePayload::Ok {
+                result: serde_json::to_value(value)?,
+            },
+        })
+    }
+
+    #[must_use]
+    pub fn err(id: u64, error: ProtoError) -> Self {
+        Self {
+            id,
+            payload: ResponsePayload::Err { error },
+        }
+    }
+}
+
+/// body of a `Response`. on the wire: a single `result` or `error` field inlined alongside `id`.
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+#[serde(untagged)]
+pub enum ResponsePayload {
+    Ok { result: Value },
+    Err { error: ProtoError },
+}
+
+/// server-to-client event on a subscribed topic.
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+pub struct Event {
+    /// e.g. `tick`, `changed`.
+    pub event: String,
+    pub topic: Topic,
+    /// shape depends on `topic` and `event`.
+    pub data: Value,
+}
+
+impl Event {
+    /// # Errors
+    /// serde error if `data` fails to serialize.
+    pub fn new<T: Serialize>(
+        topic: Topic,
+        event: impl Into<String>,
+        data: &T,
+    ) -> Result<Self, serde_json::Error> {
+        Ok(Self {
+            event: event.into(),
+            topic,
+            data: serde_json::to_value(data)?,
+        })
+    }
+}
+
+/// one frame as received from the server: paired response or fan-out event.
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+#[serde(untagged)]
+pub enum ServerFrame {
+    Response(Response),
+    Event(Event),
+}
+
+/// result body of `status`.
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+pub struct Status {
+    /// semver.
+    pub version: String,
+    pub protocol: u32,
+    pub uptime_s: u64,
+    pub profile: String,
+    pub bypass: bool,
+    /// Layer A master switch (per-app level control, global).
+    #[serde(default)]
+    pub per_app: bool,
+    pub sinks: Sinks,
+    pub streams: Vec<StreamRoute>,
+    /// per-app (Layer A) controller state; empty when not managing anything.
+    #[serde(default, skip_serializing_if = "Vec::is_empty")]
+    pub layer_a: Vec<LayerASnapshot>,
+    /// non-fatal warnings (typically profile-load: TOML parse errors, active profile missing on disk).
+    /// reflects the last successful load/reload; empty when healthy.
+    #[serde(default, skip_serializing_if = "Vec::is_empty")]
+    pub warnings: Vec<String>,
+    /// user-overlay setting overrides (dotted key → value). these shadow the active profile and
+    /// persist across `profile.use`, so a stale `headroom set` can silently override it
+    /// (e.g. `agc.enabled=false`). surfaced so clients can flag it.
+    #[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
+    pub setting_overrides: BTreeMap<String, Value>,
+}
+
+#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)]
+pub struct Sinks {
+    /// the processed virtual sink; the only sink headroom creates.
+    pub processed: SinkInfo,
+    /// hardware sink headroom forwards to, and where bypassed streams go directly.
+    pub real: SinkInfo,
+}
+
+#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)]
+pub struct SinkInfo {
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub node_id: Option<u32>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub name: Option<String>,
+    /// linked and accepting audio.
+    #[serde(default)]
+    pub ready: bool,
+    /// native rate (Hz). filter matches the *real* sink's rate to skip the output-edge resample;
+    /// the processed sink advertises whatever rate the filter currently runs at.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub sample_rate: Option<u32>,
+}
+
+/// one playback stream and its route.
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+pub struct StreamRoute {
+    pub node_id: u32,
+    /// typically `application.process.binary`.
+    pub app: String,
+    pub route: Route,
+}
+
+/// per-app (Layer A) controller state for one managed stream; on `status` and `per-app.list`.
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+pub struct LayerASnapshot {
+    pub node_id: u32,
+    pub app: String,
+    /// true while a tap + controller is actively managing the stream.
+    pub managed: bool,
+    /// last linear volume written (1.0 = unity).
+    pub volume_lin: f32,
+    /// asserted gain reduction in dB (`>= 0`; `0` = no cut).
+    pub reduction_db: f32,
+    /// user-set ceiling (linear) when ceiling-mode deference is active.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub user_ceiling_lin: Option<f32>,
+    /// strict-mode deference has locked the controller until `per-app.reset`.
+    pub deferred: bool,
+}
+
+/// entry in `profile.list`.
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+pub struct ProfileInfo {
+    pub name: String,
+    pub active: bool,
+    pub description: String,
+}
+
+/// result body of `route.list`.
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+pub struct RouteList {
+    /// in evaluation order.
+    pub rules: Vec<RouteRule>,
+    pub current: Vec<StreamRoute>,
+    /// fallback when no rule matches.
+    pub default_route: Route,
+}
+
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+pub struct RouteRule {
+    #[serde(rename = "match")]
+    pub match_: RouteRuleMatch,
+    pub route: Route,
+}
+
+/// present fields are AND'd; values within a field are OR'd; empty match is true.
+#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)]
+pub struct RouteRuleMatch {
+    #[serde(default, skip_serializing_if = "Vec::is_empty")]
+    pub process_binary: Vec<String>,
+    #[serde(default, skip_serializing_if = "Vec::is_empty")]
+    pub application_name: Vec<String>,
+    /// `pipewire.access.portal.app_id` (Flatpak).
+    #[serde(default, skip_serializing_if = "Vec::is_empty")]
+    pub portal_app_id: Vec<String>,
+    #[serde(default, skip_serializing_if = "Vec::is_empty")]
+    pub media_role: Vec<String>,
+}
+
+/// `hello` payload (sent on connect).
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+pub struct HelloData {
+    /// always `"headroom"`.
+    pub daemon: String,
+    /// semver.
+    pub version: String,
+    pub protocol: u32,
+}
+
+/// `meters` tick payload.
+#[derive(Debug, Clone, Copy, PartialEq, Serialize, Deserialize)]
+pub struct MeterTick {
+    /// BS.1770 M, 400 ms, LUFS.
+    pub momentary_lufs: f32,
+    /// BS.1770 S, 3 s, LUFS.
+    pub shortterm_lufs: f32,
+    /// BS.1770 I, gated, LUFS.
+    pub integrated_lufs: f32,
+    /// max true peak across channels, dBTP.
+    pub true_peak_dbtp: f32,
+    /// compressor + limiter, dB.
+    pub gain_reduction_db: f32,
+    pub compressor_gr_db: f32,
+    pub limiter_gr_db: f32,
+    /// AGC contribution (positive = boost), dB.
+    pub agc_gain_db: f32,
+    /// when `false`, AGC is bypassed and `agc_gain_db` is inert (0) — render as "disabled",
+    /// not an active 0 dB. absent ⇒ `true`.
+    #[serde(default = "default_true")]
+    pub agc_enabled: bool,
+    /// same as [`Self::agc_enabled`] for `compressor_gr_db`. (limiter has no flag — always-on backstop.)
+    #[serde(default = "default_true")]
+    pub compressor_enabled: bool,
+}
+
+/// absent ⇒ enabled, matching the always-on assumption before these fields existed.
+fn default_true() -> bool {
+    true
+}
+
+/// `profile` topic events.
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+#[serde(tag = "event", rename_all = "snake_case")]
+#[non_exhaustive]
+pub enum ProfileEvent {
+    Changed { name: String, previous: String },
+    /// `changed`: profiles whose definitions changed on disk.
+    Reloaded { changed: Vec<String> },
+}
+
+/// `routing` topic events.
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+#[serde(tag = "event", rename_all = "snake_case")]
+#[non_exhaustive]
+pub enum RoutingEvent {
+    StreamRouted {
+        node_id: u32,
+        app: String,
+        to: Route,
+    },
+    /// stream's PipeWire node disappeared; drop state keyed by `node_id`.
+    StreamRemoved { node_id: u32 },
+    /// Layer A tap attached; daemon now manages `Props.channelVolumes` and publishes `meters/layer_a_level`.
+    LayerAAttached { node_id: u32, app: String },
+    /// Layer A tap torn down (usually stream gone); drop Layer A state for `node_id`.
+    LayerADetached { node_id: u32 },
+    RuleChanged,
+}
+
+/// `meters/layer_a_level` payload — emitted when the Layer A controller writes a new `channelVolumes`.
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+pub struct LayerALevel {
+    pub node_id: u32,
+    pub app: String,
+    /// linear volume written (1.0 = unity).
+    pub volume_lin: f32,
+    /// asserted gain reduction, dB; ≤ 0 when reducing.
+    pub reduction_db: f32,
+}
+
+/// `daemon` topic events.
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+#[serde(tag = "event", rename_all = "snake_case")]
+#[non_exhaustive]
+pub enum DaemonEvent {
+    Started { version: String },
+    Shutdown,
+    /// events dropped on this connection.
+    Overflow {
+        lost_topic: Topic,
+        /// lost in this batch.
+        lost: u32,
+        /// total lost on this connection.
+        total_lost: u64,
+    },
+    /// non-fatal error.
+    Error {
+        /// matches `ErrorCode` when applicable.
+        code: String,
+        message: String,
+    },
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn roundtrip<T>(v: &T) -> T
+    where
+        T: Serialize + for<'de> Deserialize<'de>,
+    {
+        let s = serde_json::to_string(v).expect("serialize");
+        serde_json::from_str(&s).expect("deserialize")
+    }
+
+    #[test]
+    fn op_status_serializes_without_args() {
+        let req = Request::new(1, Op::Status);
+        let s = serde_json::to_string(&req).unwrap();
+        // Must be the flat form, no `kind` wrapper, no `args` field.
+        assert_eq!(s, r#"{"id":1,"op":"status"}"#);
+
+        let back: Request = serde_json::from_str(&s).unwrap();
+        assert_eq!(back, req);
+    }
+
+    #[test]
+    fn op_profile_use_round_trips_flat() {
+        let req = Request::new(
+            7,
+            Op::ProfileUse {
+                name: "night".into(),
+            },
+        );
+        let s = serde_json::to_string(&req).unwrap();
+        assert_eq!(s, r#"{"id":7,"op":"profile.use","args":{"name":"night"}}"#);
+
+        let back: Request = serde_json::from_str(&s).unwrap();
+        assert_eq!(back, req);
+    }
+
+    #[test]
+    fn route_set_serializes_canonical() {
+        let req = Request::new(
+            12,
+            Op::RouteSet {
+                app: "firefox".into(),
+                to: Route::Processed,
+            },
+        );
+        let s = serde_json::to_string(&req).unwrap();
+        assert_eq!(
+            s,
+            r#"{"id":12,"op":"route.set","args":{"app":"firefox","to":"processed"}}"#
+        );
+    }
+
+    #[test]
+    fn response_ok_shape() {
+        let resp = Response::ok(3, &serde_json::json!({ "name": "default" })).unwrap();
+        let s = serde_json::to_string(&resp).unwrap();
+        assert_eq!(s, r#"{"id":3,"result":{"name":"default"}}"#);
+    }
+
+    #[test]
+    fn response_err_shape() {
+        let resp = Response::err(4, ProtoError::new(crate::ErrorCode::NotFound, "missing"));
+        let s = serde_json::to_string(&resp).unwrap();
+        assert_eq!(
+            s,
+            r#"{"id":4,"error":{"code":"NOT_FOUND","message":"missing"}}"#
+        );
+    }
+
+    #[test]
+    fn server_frame_distinguishes_response_from_event() {
+        let resp = Response::ok(1, &serde_json::json!(null)).unwrap();
+        let s = serde_json::to_string(&resp).unwrap();
+        let frame: ServerFrame = serde_json::from_str(&s).unwrap();
+        assert!(matches!(frame, ServerFrame::Response(_)));
+
+        let ev = Event::new(Topic::Meters, "tick", &serde_json::json!({})).unwrap();
+        let s = serde_json::to_string(&ev).unwrap();
+        let frame: ServerFrame = serde_json::from_str(&s).unwrap();
+        assert!(matches!(frame, ServerFrame::Event(_)));
+    }
+
+    #[test]
+    fn meter_tick_roundtrip() {
+        let m = MeterTick {
+            momentary_lufs: -19.3,
+            shortterm_lufs: -20.1,
+            integrated_lufs: -19.8,
+            true_peak_dbtp: -1.4,
+            gain_reduction_db: -2.1,
+            compressor_gr_db: -0.8,
+            limiter_gr_db: -1.3,
+            agc_gain_db: 0.5,
+            agc_enabled: true,
+            compressor_enabled: true,
+        };
+        let back = roundtrip(&m);
+        assert_eq!(back, m);
+    }
+
+    #[test]
+    fn topic_string_canonical() {
+        assert_eq!(Topic::Meters.as_str(), "meters");
+        assert_eq!(serde_json::to_string(&Topic::Meters).unwrap(), "\"meters\"");
+        let t: Topic = serde_json::from_str("\"profile\"").unwrap();
+        assert_eq!(t, Topic::Profile);
+    }
+
+    #[test]
+    fn route_string_canonical() {
+        let r: Route = serde_json::from_str("\"bypass\"").unwrap();
+        assert_eq!(r, Route::Bypass);
+        assert_eq!(serde_json::to_string(&r).unwrap(), "\"bypass\"");
+    }
+
+    #[test]
+    fn error_code_screaming_snake() {
+        let s = serde_json::to_string(&crate::ErrorCode::InvalidFrame).unwrap();
+        assert_eq!(s, "\"INVALID_FRAME\"");
+        let c: crate::ErrorCode = serde_json::from_str("\"UNKNOWN_OP\"").unwrap();
+        assert_eq!(c, crate::ErrorCode::UnknownOp);
+    }
+
+    #[test]
+    fn subscribe_op_roundtrip() {
+        let req = Request::new(
+            5,
+            Op::Subscribe {
+                topics: vec![Topic::Meters, Topic::Profile],
+            },
+        );
+        let back: Request = serde_json::from_str(&serde_json::to_string(&req).unwrap()).unwrap();
+        assert_eq!(back, req);
+    }
+}

headroom

Commit dd6b8e786e1f

Commit `dd6b8e786e1f`