diff --git a/hive-priv-sock/src/framing.rs b/hive-priv-sock/src/framing.rs
new file mode 100644
index 00000000..1670e3fc
--- /dev/null
+++ b/hive-priv-sock/src/framing.rs
@@ -0,0 +1,279 @@
+//! Line framing that also carries passed file descriptors.
+//!
+//! The privileged helper speaks newline-delimited JSON. Passing a file
+//! descriptor (so a caller can hand us a connected socket to write a
+//! `btrfs send` stream into, without us ever learning what it connects
+//! to) means `recvmsg` instead of a plain `read`, because `SCM_RIGHTS`
+//! ancillary data is attached to one specific `recvmsg` call and a
+//! buffered line reader cannot surface it.
+//!
+//! It lives in the protocol crate rather than in `hive-priv` because
+//! both ends need it: the helper reassembles what the daemon sends, so
+//! framing is part of the wire contract, not an implementation detail
+//! of one side.
+//!
+//! Everything subtle about that lives here, deliberately kept free of
+//! syscalls so it can be tested exhaustively: the caller does the
+//! `recvmsg` and feeds us `(bytes, fds)`; we hand back complete
+//! messages with the descriptor that belongs to each. See [`Framer`]
+//! for why that pairing is not the obvious one.
+use std::collections::VecDeque;
+use std::os::fd::OwnedFd;
+
+/// One complete request line plus the descriptor it claimed, if any.
+pub struct Framed {
+    /// The message body — a JSON line, without its trailing newline.
+    pub line: String,
+    /// The descriptor this message claimed, oldest-first. `None` when
+    /// no unclaimed descriptor was pending, which is the ordinary case
+    /// for every request that doesn't pass one.
+    pub fd: Option<OwnedFd>,
+}
+
+/// Reassembles `(bytes, fds)` chunks into complete messages.
+///
+/// Bytes accumulate until a newline; descriptors queue independently.
+/// A message claims the oldest queued descriptor at the moment the
+/// message *completes*, never at the moment a chunk arrives.
+///
+/// That indirection is the point. A descriptor does *not* arrive
+/// neatly paired with the request that wants it — it can arrive with a
+/// chunk holding only part of its request's line, with a chunk whose
+/// bytes finish the *previous* request, ahead of any of its own bytes,
+/// or alongside several complete requests at once.
+///
+/// So pairing "the fd from this chunk" with "the request in this
+/// chunk" is wrong, and wrong in the worst way: the types are
+/// identical either way, so nothing catches it, and the failure is one
+/// request executing against another's descriptor — here, writing one
+/// agent's state into a different transfer's socket.
+#[derive(Default)]
+pub struct Framer {
+    buf: Vec<u8>,
+    pending_fds: VecDeque<OwnedFd>,
+}
+
+impl Framer {
+    /// Feed one `recvmsg` result: the bytes it read and any descriptors
+    /// it carried.
+    ///
+    /// Returns every message completed by this chunk, in order. A chunk
+    /// can complete several (or none).
+    ///
+    /// Invalid UTF-8 in a line is surfaced as an error rather than
+    /// lossily replaced: the body is JSON, and a mangled line should
+    /// fail loudly instead of parsing into something adjacent.
+    pub fn push(&mut self, bytes: &[u8], fds: Vec<OwnedFd>) -> Vec<Result<Framed, FramingError>> {
+        self.pending_fds.extend(fds);
+        self.buf.extend_from_slice(bytes);
+
+        let mut out = Vec::new();
+        while let Some(nl) = self.buf.iter().position(|&b| b == b'\n') {
+            let line_bytes: Vec<u8> = self.buf.drain(..=nl).take(nl).collect();
+            match String::from_utf8(line_bytes) {
+                Ok(line) => out.push(Ok(Framed {
+                    line,
+                    fd: self.pending_fds.pop_front(),
+                })),
+                // The descriptor is deliberately NOT claimed for a line
+                // we can't read: dropping it here would close a
+                // descriptor that belongs to a request we never
+                // processed. It stays queued for the next complete
+                // message, and `drain_pending_fds` closes whatever is
+                // left when the connection ends.
+                Err(e) => out.push(Err(FramingError::InvalidUtf8(e.utf8_error().to_string()))),
+            }
+        }
+        out
+    }
+
+    /// How many descriptors have arrived but not yet been claimed.
+    ///
+    /// Exposed for the connection teardown path: anything still here
+    /// when a connection ends belongs to no request and must be closed,
+    /// or a long-lived helper leaks a descriptor per abandoned message.
+    pub fn pending_fd_count(&self) -> usize {
+        self.pending_fds.len()
+    }
+
+    /// Take every unclaimed descriptor so the caller can drop (close)
+    /// them explicitly at teardown.
+    pub fn drain_pending_fds(&mut self) -> Vec<OwnedFd> {
+        self.pending_fds.drain(..).collect()
+    }
+
+    /// Bytes buffered without a terminating newline yet.
+    ///
+    /// A connection that ends here sent a partial request; the caller
+    /// should treat that as a truncated message rather than silently
+    /// discarding it.
+    pub fn partial_len(&self) -> usize {
+        self.buf.len()
+    }
+}
+
+/// A message that could not be framed.
+#[derive(Debug)]
+pub enum FramingError {
+    /// The line was not valid UTF-8. Carries the underlying reason as
+    /// a string so the error type stays trivially cloneable/loggable.
+    InvalidUtf8(String),
+}
+
+impl std::fmt::Display for FramingError {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        match self {
+            Self::InvalidUtf8(e) => write!(f, "request line was not valid UTF-8: {e}"),
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use std::os::fd::{AsRawFd as _, OwnedFd};
+
+    /// A real, owned descriptor to pass around. `/dev/null` is opened
+    /// rather than faked so the tests exercise genuine `OwnedFd` move
+    /// semantics — a fake would not catch a double-close.
+    fn some_fd() -> OwnedFd {
+        std::fs::File::open("/dev/null")
+            .expect("open /dev/null")
+            .into()
+    }
+
+    fn ok(res: Result<Framed, FramingError>) -> Framed {
+        res.expect("expected a framed message")
+    }
+
+    #[test]
+    fn line_without_fd_claims_nothing() {
+        let mut f = Framer::default();
+        let out = f.push(b"{\"a\":1}\n", vec![]);
+        assert_eq!(out.len(), 1);
+        let m = ok(out.into_iter().next().unwrap());
+        assert_eq!(m.line, "{\"a\":1}");
+        assert!(
+            m.fd.is_none(),
+            "no descriptor was sent, none should be claimed"
+        );
+    }
+
+    #[test]
+    fn fd_arriving_with_a_partial_line_is_claimed_by_that_line() {
+        // The case the whole type exists for: the descriptor rides in
+        // on a chunk that does not complete any message.
+        let mut f = Framer::default();
+        let raw = {
+            let fd = some_fd();
+            let raw = fd.as_raw_fd();
+            let out = f.push(b"{\"partial\"", vec![fd]);
+            assert!(out.is_empty(), "no newline yet, so no message yet");
+            raw
+        };
+        assert_eq!(f.pending_fd_count(), 1, "descriptor waits for its message");
+
+        let out = f.push(b":true}\n", vec![]);
+        assert_eq!(out.len(), 1);
+        let m = ok(out.into_iter().next().unwrap());
+        assert_eq!(m.line, "{\"partial\":true}");
+        assert_eq!(
+            m.fd.expect("the completed message must claim it")
+                .as_raw_fd(),
+            raw
+        );
+    }
+
+    #[test]
+    fn fd_does_not_leak_to_a_neighbouring_message_in_the_same_chunk() {
+        // Two complete messages arrive together with ONE descriptor.
+        // It belongs to the first; the second must get nothing rather
+        // than inheriting it.
+        let mut f = Framer::default();
+        let out = f.push(b"{\"first\":1}\n{\"second\":2}\n", vec![some_fd()]);
+        assert_eq!(out.len(), 2);
+        let mut it = out.into_iter();
+        let first = ok(it.next().unwrap());
+        let second = ok(it.next().unwrap());
+        assert_eq!(first.line, "{\"first\":1}");
+        assert!(first.fd.is_some(), "oldest message claims the descriptor");
+        assert_eq!(second.line, "{\"second\":2}");
+        assert!(
+            second.fd.is_none(),
+            "a second message must not inherit the first's descriptor"
+        );
+    }
+
+    #[test]
+    fn descriptors_are_claimed_oldest_first() {
+        let mut f = Framer::default();
+        let (a, b) = (some_fd(), some_fd());
+        let (ra, rb) = (a.as_raw_fd(), b.as_raw_fd());
+        f.push(b"", vec![a]);
+        f.push(b"", vec![b]);
+        let out = f.push(b"one\ntwo\n", vec![]);
+        assert_eq!(out.len(), 2);
+        let mut it = out.into_iter();
+        assert_eq!(ok(it.next().unwrap()).fd.unwrap().as_raw_fd(), ra);
+        assert_eq!(ok(it.next().unwrap()).fd.unwrap().as_raw_fd(), rb);
+    }
+
+    #[test]
+    fn a_chunk_may_complete_nothing() {
+        let mut f = Framer::default();
+        assert!(f.push(b"no newline here", vec![]).is_empty());
+        assert_eq!(
+            f.partial_len(),
+            15,
+            "bytes stay buffered for the next chunk"
+        );
+    }
+
+    #[test]
+    fn invalid_utf8_errors_without_consuming_a_descriptor() {
+        // A line we cannot read must not swallow a descriptor: closing
+        // it here would destroy something belonging to a request that
+        // was never processed.
+        let mut f = Framer::default();
+        let out = f.push(b"\xff\xfe\n", vec![some_fd()]);
+        assert_eq!(out.len(), 1);
+        assert!(matches!(
+            out.into_iter().next().unwrap(),
+            Err(FramingError::InvalidUtf8(_))
+        ));
+        assert_eq!(
+            f.pending_fd_count(),
+            1,
+            "the descriptor must survive a bad line, not be consumed by it"
+        );
+    }
+
+    #[test]
+    fn unclaimed_descriptors_are_recoverable_for_teardown() {
+        // Whatever is still queued when a connection ends belongs to no
+        // request. The caller has to be able to get it back and close
+        // it, or a long-lived helper leaks one per abandoned message.
+        let mut f = Framer::default();
+        f.push(b"", vec![some_fd(), some_fd()]);
+        assert_eq!(f.pending_fd_count(), 2);
+        assert_eq!(f.drain_pending_fds().len(), 2);
+        assert_eq!(f.pending_fd_count(), 0);
+    }
+
+    #[test]
+    fn empty_line_is_a_message_not_a_skip() {
+        // A bare newline is a zero-length request. It should surface as
+        // an (invalid) message for the dispatcher to reject, rather
+        // than being silently dropped here — silently dropping it would
+        // desynchronise descriptor claiming.
+        let mut f = Framer::default();
+        let fd_present = {
+            let out = f.push(b"\n", vec![some_fd()]);
+            assert_eq!(out.len(), 1);
+            let m = ok(out.into_iter().next().unwrap());
+            assert_eq!(m.line, "");
+            m.fd.is_some()
+        };
+        assert!(fd_present, "an empty line still claims its descriptor");
+    }
+}
diff --git a/hive-priv-sock/src/lib.rs b/hive-priv-sock/src/lib.rs
index fcb71318..32c85fe0 100644
--- a/hive-priv-sock/src/lib.rs
+++ b/hive-priv-sock/src/lib.rs
@@ -12,6 +12,12 @@
 
 use serde::{Deserialize, Serialize};
 
+// Line framing that also carries passed file descriptors. Part of the
+// wire contract rather than one side's implementation detail: the
+// daemon sends descriptors and the helper reassembles them, so both
+// ends have to agree on how a descriptor is bound to a request.
+pub mod framing;
+
 /// Default socket path for the privileged helper.
 pub const PRIV_SOCK: &str = "/run/hive/priv.sock";