Implement Cast Streaming mirroring, DLNA casting, daemon+GUI, and breadd integration

Builds out the full v1 scope: a vendored+patched openscreen subset for
low-latency Cast Streaming (Mirroring receiver 0F5096E8) alongside the
existing Cast V2/HLS and new DLNA/AVTransport casting paths, breadcastd's
Idle/Casting state machine with a private IPC socket, the breadcast GTK4
popup as a thin IPC client, and bread.cast.*/bread.command.cast.* breadd
integration (device discovery, start/stop, mirroring lifecycle events).
Also adds bakery/systemd/Forgejo CI packaging.

Validated end-to-end against a real Chromecast/Google TV: negotiated
Cast Streaming session, live pipeline playback, and daemon+GUI click-to-cast/
stop through the actual popup.

breadcastd/Cargo.toml

@@ -0,0 +1,24 @@
+[package]
+name = "breadcastd"
+version.workspace = true
+edition.workspace = true
+license.workspace = true
+authors.workspace = true
+description = "breadcast background daemon: discovery, capture/encode/serve pipeline, Cast V2 session"
+
+[[bin]]
+name = "breadcastd"
+path = "src/main.rs"
+
+[dependencies]
+breadcast-core = { path = "../breadcast-core" }
+anyhow = { workspace = true }
+serde = { workspace = true }
+tracing = { workspace = true }
+tracing-subscriber = { workspace = true }
+tokio = { workspace = true }
+serde_json = { workspace = true }
+rust_cast = { version = "0.21", features = ["thread_safe"] }
+gstreamer = "0.25"
+gstreamer-app = "0.25"
+bread-utils = { git = "https://git.breadway.dev/Breadway/bread-ecosystem", tag = "v0.7.1", features = ["bread-client"] }

breadcastd/src/bread_events.rs

@@ -0,0 +1,93 @@
+//! `bread.cast.*` event integration — optional, non-blocking. See
+//! `EVENTS.md` at the repo root for the full contract. breadcastd works
+//! identically with or without breadd running; every call here is
+//! fire-and-forget (`BreadClient::emit` never blocks or errors this
+//! process) so a missing or restarting breadd never affects discovery or
+//! mirroring itself.
+
+use bread_utils::bread_client::{BreadClient, BreadEvent};
+use tokio::sync::{mpsc, oneshot};
+
+use crate::daemon::DaemonCommand;
+
+/// This app's id in bread's sibling-app namespace registry
+/// (`bread_shared::apps::KNOWN_APPS`) — events publish as `bread.cast.*`,
+/// commands arrive on `bread.command.cast.*`.
+pub const APP_ID: &str = "cast";
+
+pub fn emit_device_found(client: &BreadClient, id: &str, name: &str, model: &str, protocol: &str) {
+    client.emit(
+        "bread.cast.device_found",
+        serde_json::json!({
+            "id": id,
+            "name": name,
+            "model": model,
+            "protocol": protocol,
+        }),
+    );
+}
+
+pub fn emit_mirroring_started(client: &BreadClient, device_id: &str, device_name: &str, protocol: &str) {
+    client.emit(
+        "bread.cast.mirroring_started",
+        serde_json::json!({
+            "device_id": device_id,
+            "device_name": device_name,
+            "protocol": protocol,
+        }),
+    );
+}
+
+pub fn emit_mirroring_stopped(client: &BreadClient) {
+    client.emit("bread.cast.mirroring_stopped", serde_json::json!({}));
+}
+
+pub fn emit_mirroring_failed(client: &BreadClient, device_id: &str, error: &str) {
+    client.emit(
+        "bread.cast.mirroring_failed",
+        serde_json::json!({
+            "device_id": device_id,
+            "error": error,
+        }),
+    );
+}
+
+/// Reacts to `bread.command.cast.*` verbs, e.g. from a Hyprland keybind.
+/// Runs on `BreadClient::subscribe`'s dedicated background thread (a plain
+/// `std::thread`, not a tokio worker) — `mpsc::Sender::blocking_send` is
+/// safe to call from here for exactly that reason, but would panic if
+/// called from inside the tokio runtime.
+///
+/// Both verbs reply is intentionally not awaited: the reply channel exists
+/// because `DaemonCommand::StartCast`/`StopCast` need one for the IPC
+/// socket's request/response use (see `ipc.rs`), but a fire-and-forget bus
+/// command has nowhere to deliver a reply to anyway — the outcome shows up
+/// as a `bread.cast.mirroring_started`/`.failed`/`.stopped` event instead
+/// (see `daemon.rs`'s `start_cast`/`StopCast` handling).
+pub fn handle_command(event: &BreadEvent, daemon_tx: &mpsc::Sender<DaemonCommand>) {
+    let Some(verb) = event.event.strip_prefix("bread.command.cast.") else {
+        return;
+    };
+    match verb {
+        "start" => {
+            let Some(device_id) = event.data.get("device_id").and_then(|v| v.as_str()) else {
+                tracing::warn!("bread.command.cast.start missing a string \"device_id\", ignoring");
+                return;
+            };
+            let (reply, _reply_rx) = oneshot::channel();
+            if daemon_tx
+                .blocking_send(DaemonCommand::StartCast { device_id: device_id.to_string(), reply })
+                .is_err()
+            {
+                tracing::warn!("daemon actor unavailable, dropping bread.command.cast.start");
+            }
+        }
+        "stop" => {
+            let (reply, _reply_rx) = oneshot::channel();
+            if daemon_tx.blocking_send(DaemonCommand::StopCast { reply }).is_err() {
+                tracing::warn!("daemon actor unavailable, dropping bread.command.cast.stop");
+            }
+        }
+        other => tracing::info!(verb = other, "ignoring unknown bread.command.cast verb"),
+    }
+}

breadcastd/src/cast_mirror.rs

@@ -0,0 +1,202 @@
+//! Owns one active Cast Streaming mirroring session end-to-end: portal
+//! capture, the GStreamer encode pipeline, the CASTV2 connection to the
+//! Mirroring receiver, and the three pump threads that shuttle
+//! OFFER/ANSWER messages and encoded frames between them. This is
+//! `cast_stream_test.rs`'s orchestration, restructured into something the
+//! daemon can start and stop on demand instead of running for a fixed
+//! duration from a CLI `main`.
+//!
+//! The Cast Streaming (low-latency, RTP-based) path — see `dlna_mirror.rs`
+//! for the DLNA/UPnP counterpart (HLS-over-HTTP, polled instead of pushed).
+
+use std::sync::Arc;
+use std::sync::atomic::{AtomicBool, Ordering};
+
+use anyhow::{Context, Result};
+use breadcast_core::caststream::{CastStreamEvent, VideoParams, WEBRTC_NAMESPACE};
+use breadcast_core::pipeline::{
+    build_video_pipeline_for_streaming, pull_encoded_frame, request_key_frame, set_video_bitrate_kbps,
+};
+use breadcast_core::{CastDevice, CaptureSession, CastSession, CastStreamSender};
+use gstreamer as gst;
+use gstreamer::prelude::*;
+use rust_cast::channels::receiver::CastDeviceApp;
+
+use crate::daemon::DaemonCommand;
+
+pub struct CastMirrorSession {
+    pipeline: gst::Pipeline,
+    session: CastSession,
+    capture: Option<CaptureSession>,
+    threads: Vec<std::thread::JoinHandle<()>>,
+}
+
+impl CastMirrorSession {
+    /// Starts mirroring to `device`. Blocks (briefly) on the portal picker,
+    /// the CASTV2 handshake, and OFFER/ANSWER negotiation before returning
+    /// -- by the time this resolves, frames are already flowing.
+    ///
+    /// `daemon_tx` is used to report unprompted session death (a GStreamer
+    /// error, the user clicking "stop sharing" in the portal picker, the
+    /// receiver dropping the connection) back to the daemon actor, so it
+    /// can transition back to `Idle` and notify GUI clients even if nobody
+    /// called `stop()`.
+    pub async fn start(device: CastDevice, daemon_tx: tokio::sync::mpsc::Sender<DaemonCommand>) -> Result<Self> {
+        let capture = CaptureSession::start().await.context("failed to start portal screen capture")?;
+        let video_node_id = capture.video_node_id();
+
+        let (pipeline, appsink, encoder) =
+            build_video_pipeline_for_streaming(video_node_id).context("failed to build the encode pipeline")?;
+
+        {
+            let pipeline_watch = pipeline.clone();
+            std::thread::spawn(move || {
+                match breadcast_core::pipeline::run_until_error_or_timeout(&pipeline_watch, gst::ClockTime::from_seconds(3600))
+                {
+                    Ok(outcome) => tracing::debug!(?outcome, "encode pipeline bus watcher ended"),
+                    Err(e) => tracing::error!(error = ?e, "encode pipeline error"),
+                }
+            });
+        }
+
+        // The blocking CASTV2 TCP+TLS handshake + app launch is quick
+        // (milliseconds on a LAN) but still blocking I/O -- run it off the
+        // async worker thread pool rather than stalling it, even briefly.
+        let device_for_connect = device.clone();
+        let (session, _media_events, raw_messages) = tokio::task::spawn_blocking(move || {
+            CastSession::connect_app(
+                &device_for_connect,
+                CastDeviceApp::Custom(breadcast_core::caststream::MIRRORING_APP_ID.to_string()),
+            )
+        })
+        .await
+        .context("connect_app task panicked")?
+        .context("failed to connect and launch the Mirroring receiver")?;
+
+        let (sender, stream_events) =
+            CastStreamSender::start(&device.host, "sender-0", session.transport_id(), VideoParams::default())
+                .context("failed to start the Cast Streaming session")?;
+        let sender = Arc::new(sender);
+
+        let mut threads = Vec::new();
+
+        threads.push({
+            let sender = sender.clone();
+            std::thread::spawn(move || {
+                while let Some(msg) = raw_messages.recv() {
+                    if msg.namespace == WEBRTC_NAMESPACE {
+                        sender.on_message(&msg.source_id, &msg.namespace, &msg.message);
+                    }
+                }
+            })
+        });
+
+        let negotiated = Arc::new(AtomicBool::new(false));
+        threads.push({
+            let session = session.clone();
+            let negotiated = negotiated.clone();
+            std::thread::spawn(move || {
+                while let Ok(event) = stream_events.recv() {
+                    match event {
+                        CastStreamEvent::OutboundMessage { message, .. } => {
+                            if let Err(e) = session.send_raw_message(WEBRTC_NAMESPACE, &message) {
+                                tracing::warn!(error = ?e, "failed to send Cast Streaming message");
+                            }
+                        }
+                        CastStreamEvent::Negotiated => negotiated.store(true, Ordering::Release),
+                        CastStreamEvent::Error(message) => tracing::warn!(%message, "Cast Streaming error"),
+                        CastStreamEvent::PictureLost => tracing::debug!("receiver reported picture loss"),
+                    }
+                }
+            })
+        });
+
+        tracing::info!(device = %device.name, "sending Cast Streaming OFFER");
+        sender.negotiate();
+
+        let deadline = tokio::time::Instant::now() + tokio::time::Duration::from_secs(10);
+        while !negotiated.load(Ordering::Acquire) && tokio::time::Instant::now() < deadline {
+            tokio::time::sleep(tokio::time::Duration::from_millis(50)).await;
+        }
+        if !negotiated.load(Ordering::Acquire) {
+            let _ = session.stop();
+            capture.close().await.ok();
+            anyhow::bail!("never received an ANSWER from {} (negotiation timed out)", device.name);
+        }
+
+        pipeline.set_state(gst::State::Playing).context("failed to start the encode pipeline")?;
+        tracing::info!(device = %device.name, "mirroring started");
+
+        threads.push({
+            let device_name = device.name.clone();
+            std::thread::spawn(move || {
+                let result = frame_pump_loop(&appsink, &encoder, &sender);
+                if let Err(e) = result {
+                    tracing::warn!(device = %device_name, error = ?e, "frame pump ended with an error");
+                }
+                // Best-effort: if this is running, the daemon actor is (or
+                // was, very recently) still alive. If the channel is full or
+                // closed, there's nothing more useful to do from this
+                // thread than drop the notification.
+                let _ = daemon_tx.blocking_send(DaemonCommand::SessionEnded);
+            })
+        });
+
+        Ok(Self { pipeline, session, capture: Some(capture), threads })
+    }
+
+    /// Tears down the session: stops the pipeline (which unblocks the frame
+    /// pump thread's blocking `appsink.pull_sample()` call), stops the
+    /// CASTV2 session (which ends its io thread, closing the channels the
+    /// other two pump threads block on), then joins every thread.
+    pub async fn stop(mut self) {
+        if let Err(e) = self.pipeline.set_state(gst::State::Null) {
+            tracing::warn!(error = ?e, "failed to stop the encode pipeline cleanly");
+        }
+        if let Err(e) = self.session.stop() {
+            tracing::warn!(error = ?e, "failed to cleanly stop the cast session");
+        }
+        if let Some(capture) = self.capture.take() {
+            if let Err(e) = capture.close().await {
+                tracing::warn!(error = ?e, "failed to cleanly close the portal capture session");
+            }
+        }
+        for thread in self.threads.drain(..) {
+            // These threads all end once the pipeline/session teardown
+            // above propagates to them (see this method's own doc comment)
+            // -- `spawn_blocking` just keeps `.join()`'s wait off the async
+            // runtime's worker threads.
+            if let Err(panic) = tokio::task::spawn_blocking(move || thread.join()).await {
+                tracing::warn!(error = ?panic, "mirror session pump thread join task panicked");
+            }
+        }
+    }
+}
+
+fn frame_pump_loop(
+    appsink: &gstreamer_app::AppSink,
+    encoder: &gst::Element,
+    sender: &CastStreamSender,
+) -> Result<()> {
+    let mut last_bitrate_update = std::time::Instant::now();
+    loop {
+        let Some((data, is_key_frame, capture_time_us)) = pull_encoded_frame(appsink)? else {
+            return Ok(()); // EOS -- pipeline was set to Null, or the portal source ended
+        };
+
+        if sender.needs_key_frame() && !is_key_frame {
+            request_key_frame(appsink);
+        }
+
+        if let Err(e) = sender.enqueue_frame(&data, is_key_frame, capture_time_us) {
+            tracing::debug!(error = ?e, "dropped a frame (not negotiated yet or backpressure)");
+        }
+
+        if last_bitrate_update.elapsed() >= std::time::Duration::from_secs(1) {
+            let bps = sender.estimated_bandwidth_bps();
+            let target_kbps = ((bps as f64 * 0.85) / 1000.0).max(500.0) as u32;
+            set_video_bitrate_kbps(encoder, target_kbps);
+            last_bitrate_update = std::time::Instant::now();
+        }
+    }
+}

breadcastd/src/daemon.rs

@@ -0,0 +1,221 @@
+//! The daemon's single state-owning actor: current `StateInfo`, the known
+//! Cast and DLNA device lists, and the active mirroring session (if any)
+//! all live here, touched only by this task -- the same "single owner +
+//! channel" pattern `breadcast-core::cast_sender::CastSession` uses for its
+//! io thread, for the same reason (avoids retrofitting locks around state
+//! that multiple IPC connections and two independent discovery loops all
+//! need to touch).
+
+use std::collections::HashMap;
+
+use bread_utils::bread_client::BreadClient;
+use breadcast_core::ipc::{DeviceInfo, Protocol, ServerMessage, StateInfo};
+use breadcast_core::{CastDevice, DlnaDevice};
+use tokio::sync::{broadcast, mpsc, oneshot};
+
+use crate::bread_events;
+use crate::cast_mirror::CastMirrorSession;
+use crate::dlna_mirror::DlnaMirrorSession;
+
+pub enum DaemonCommand {
+    ListDevices { reply: oneshot::Sender<Vec<DeviceInfo>> },
+    GetState { reply: oneshot::Sender<StateInfo> },
+    StartCast { device_id: String, reply: oneshot::Sender<Result<(), String>> },
+    StopCast { reply: oneshot::Sender<Result<(), String>> },
+    CastDeviceFound(CastDevice),
+    CastDeviceLost(String),
+    DlnaDeviceFound(DlnaDevice),
+    /// DLNA devices have no separate stable id -- their description URL
+    /// (`DlnaDevice::url`) doubles as one, see `DlnaDevice`'s doc comment.
+    DlnaDeviceLost(String),
+    /// Sent by an active session's own background thread/task when it ends
+    /// on its own (portal "stop sharing", a GStreamer error, the receiver
+    /// dropping the connection or stopping playback) -- as opposed to
+    /// `StopCast` being called. Either way the daemon needs to forget the
+    /// (now-dead) session and go back to `Idle`.
+    SessionEnded,
+}
+
+pub fn spawn(events_tx: broadcast::Sender<ServerMessage>, bread_client: BreadClient) -> mpsc::Sender<DaemonCommand> {
+    let (command_tx, mut command_rx) = mpsc::channel(32);
+    let self_tx = command_tx.clone();
+    tokio::spawn(async move {
+        let mut daemon = Daemon {
+            cast_devices: HashMap::new(),
+            dlna_devices: HashMap::new(),
+            state: StateInfo::Idle,
+            active_session: None,
+            events_tx,
+            bread_client,
+            self_tx,
+        };
+        while let Some(command) = command_rx.recv().await {
+            daemon.handle(command).await;
+        }
+    });
+    command_tx
+}
+
+/// The currently active mirroring session, if any -- exactly one of the two
+/// protocol-specific session types, chosen by which device map `StartCast`
+/// found the requested device id in.
+enum ActiveSession {
+    Cast(CastMirrorSession),
+    Dlna(Box<DlnaMirrorSession>),
+}
+
+impl ActiveSession {
+    async fn stop(self) {
+        match self {
+            ActiveSession::Cast(session) => session.stop().await,
+            ActiveSession::Dlna(session) => session.stop().await,
+        }
+    }
+}
+
+struct Daemon {
+    cast_devices: HashMap<String, CastDevice>,
+    /// Keyed by `DlnaDevice::url`, the closest thing DLNA has to a stable
+    /// device id -- see `breadcast_core::ipc::DeviceInfo`'s doc comment.
+    dlna_devices: HashMap<String, DlnaDevice>,
+    state: StateInfo,
+    active_session: Option<ActiveSession>,
+    events_tx: broadcast::Sender<ServerMessage>,
+    /// Used to publish `bread.cast.mirroring_started`/`.stopped`/`.failed`
+    /// on state transitions -- see `bread_events.rs`. A no-op if breadd
+    /// isn't running (see that module's doc comment).
+    bread_client: BreadClient,
+    /// A clone of this actor's own command sender, handed to each mirror
+    /// session so its background pump/poll can report unprompted death
+    /// (see `DaemonCommand::SessionEnded`) without this module needing to
+    /// expose anything beyond the command channel itself.
+    self_tx: mpsc::Sender<DaemonCommand>,
+}
+
+impl Daemon {
+    async fn handle(&mut self, command: DaemonCommand) {
+        match command {
+            DaemonCommand::ListDevices { reply } => {
+                let _ = reply.send(self.device_list());
+            }
+            DaemonCommand::GetState { reply } => {
+                let _ = reply.send(self.state.clone());
+            }
+            DaemonCommand::StartCast { device_id, reply } => {
+                self.start_cast(device_id, reply).await;
+            }
+            DaemonCommand::StopCast { reply } => {
+                if let Some(session) = self.active_session.take() {
+                    session.stop().await;
+                    bread_events::emit_mirroring_stopped(&self.bread_client);
+                }
+                self.state = StateInfo::Idle;
+                self.broadcast_state();
+                let _ = reply.send(Ok(()));
+            }
+            DaemonCommand::SessionEnded => {
+                // The session already tore itself down (that's what
+                // triggered this) -- just drop our handle to it and update
+                // state. Ignored if this arrives after an explicit
+                // `StopCast` already cleared `active_session` (the
+                // background pump/poll it came from may briefly outlive
+                // that call).
+                if self.active_session.take().is_some() {
+                    tracing::info!("mirror session ended on its own, returning to idle");
+                    self.state = StateInfo::Idle;
+                    self.broadcast_state();
+                    bread_events::emit_mirroring_stopped(&self.bread_client);
+                }
+            }
+            DaemonCommand::CastDeviceFound(device) => {
+                self.cast_devices.insert(device.id.clone(), device);
+                self.broadcast_devices();
+            }
+            DaemonCommand::CastDeviceLost(id) => {
+                self.cast_devices.remove(&id);
+                self.broadcast_devices();
+            }
+            DaemonCommand::DlnaDeviceFound(device) => {
+                self.dlna_devices.insert(device.url.clone(), device);
+                self.broadcast_devices();
+            }
+            DaemonCommand::DlnaDeviceLost(url) => {
+                self.dlna_devices.remove(&url);
+                self.broadcast_devices();
+            }
+        }
+    }
+
+    async fn start_cast(&mut self, device_id: String, reply: oneshot::Sender<Result<(), String>>) {
+        if self.active_session.is_some() {
+            let _ = reply.send(Err("already casting -- stop the current session first".to_string()));
+            return;
+        }
+
+        if let Some(device) = self.cast_devices.get(&device_id).cloned() {
+            match CastMirrorSession::start(device.clone(), self.self_tx.clone()).await {
+                Ok(session) => {
+                    self.active_session = Some(ActiveSession::Cast(session));
+                    self.state =
+                        StateInfo::Casting { device_id: device.id.clone(), device_name: device.name.clone(), protocol: Protocol::Cast };
+                    self.broadcast_state();
+                    bread_events::emit_mirroring_started(&self.bread_client, &device.id, &device.name, "cast");
+                    let _ = reply.send(Ok(()));
+                }
+                Err(e) => {
+                    bread_events::emit_mirroring_failed(&self.bread_client, &device.id, &e.to_string());
+                    let _ = reply.send(Err(e.to_string()));
+                }
+            }
+            return;
+        }
+
+        if let Some(device) = self.dlna_devices.get(&device_id).cloned() {
+            match DlnaMirrorSession::start(device.clone(), self.self_tx.clone()).await {
+                Ok(session) => {
+                    self.active_session = Some(ActiveSession::Dlna(Box::new(session)));
+                    self.state = StateInfo::Casting {
+                        device_id: device.url.clone(),
+                        device_name: device.friendly_name.clone(),
+                        protocol: Protocol::Dlna,
+                    };
+                    self.broadcast_state();
+                    bread_events::emit_mirroring_started(&self.bread_client, &device.url, &device.friendly_name, "dlna");
+                    let _ = reply.send(Ok(()));
+                }
+                Err(e) => {
+                    bread_events::emit_mirroring_failed(&self.bread_client, &device.url, &e.to_string());
+                    let _ = reply.send(Err(e.to_string()));
+                }
+            }
+            return;
+        }
+
+        let _ = reply.send(Err(format!("unknown device id \"{device_id}\"")));
+    }
+
+    fn device_list(&self) -> Vec<DeviceInfo> {
+        let mut devices: Vec<DeviceInfo> = self
+            .cast_devices
+            .values()
+            .map(|d| DeviceInfo { id: d.id.clone(), name: d.name.clone(), model: d.model.clone(), protocol: Protocol::Cast })
+            .collect();
+        devices.extend(self.dlna_devices.values().map(|d| DeviceInfo {
+            id: d.url.clone(),
+            name: d.friendly_name.clone(),
+            model: "DLNA renderer".to_string(),
+            protocol: Protocol::Dlna,
+        }));
+        devices
+    }
+
+    fn broadcast_state(&self) {
+        let data = serde_json::to_value(&self.state).expect("StateInfo always serializes");
+        let _ = self.events_tx.send(ServerMessage::Event { event: "state_changed".to_string(), data });
+    }
+
+    fn broadcast_devices(&self) {
+        let data = serde_json::to_value(self.device_list()).expect("Vec<DeviceInfo> always serializes");
+        let _ = self.events_tx.send(ServerMessage::Event { event: "device_list_changed".to_string(), data });
+    }
+}

breadcastd/src/dlna_mirror.rs

@@ -0,0 +1,139 @@
+//! Owns one active DLNA/UPnP mirroring session: portal capture, the
+//! GStreamer HLS encode pipeline, the local HTTP server, and the
+//! `AVTransport` control session. This is `dlna_mirror_test.rs`'s
+//! orchestration, restructured into something the daemon can start and
+//! stop on demand — the DLNA counterpart to `cast_mirror.rs`'s
+//! `CastMirrorSession`.
+//!
+//! Unlike the Cast Streaming path, there's no persistent bidirectional
+//! connection to a DLNA renderer to read events from — `AVTransport` is a
+//! plain SOAP request/response protocol (see `DlnaSession`'s doc comment),
+//! so "did the renderer stop on its own" can only be *polled*, not pushed.
+
+use std::time::Duration;
+
+use anyhow::{Context, Result};
+use breadcast_core::net::local_lan_ip;
+use breadcast_core::pipeline::{build_video_pipeline, hls_output_dir, wait_for_playlist_segments};
+use breadcast_core::{CaptureSession, DlnaDevice, DlnaSession};
+use gstreamer as gst;
+use gstreamer::prelude::*;
+
+use crate::daemon::DaemonCommand;
+
+/// How often to poll `GetTransportInfo` for an unprompted stop (the user
+/// stopped playback from the TV's own remote, or the renderer just dropped
+/// the stream) — see [`DlnaSession::transport_state`]'s doc comment for why
+/// this is a poll, not a push.
+const POLL_INTERVAL: Duration = Duration::from_secs(3);
+
+pub struct DlnaMirrorSession {
+    pipeline: gst::Pipeline,
+    session: DlnaSession,
+    capture: Option<CaptureSession>,
+    poll_task: tokio::task::JoinHandle<()>,
+}
+
+impl DlnaMirrorSession {
+    /// Starts mirroring to `device`. Blocks (briefly) on the portal picker,
+    /// pipeline startup, and the renderer accepting the `SetAVTransportURI`
+    /// + `Play` actions before returning.
+    ///
+    /// `daemon_tx` is used to report an unprompted session end (the
+    /// renderer stopping playback on its own, a GStreamer error, or the
+    /// renderer becoming unreachable) back to the daemon actor — mirrors
+    /// `CastMirrorSession::start`'s same use of it.
+    pub async fn start(device: DlnaDevice, daemon_tx: tokio::sync::mpsc::Sender<DaemonCommand>) -> Result<Self> {
+        let capture = CaptureSession::start().await.context("failed to start portal screen capture")?;
+        let video_node_id = capture.video_node_id();
+
+        let output_dir = hls_output_dir("dlna-mirror")?;
+        let pipeline =
+            build_video_pipeline(video_node_id, &output_dir).context("failed to build the encode pipeline")?;
+
+        // Fire-and-forget, same as `CastMirrorSession::start`'s identical
+        // block: nothing joins this thread, it just self-terminates on
+        // pipeline error, EOS, or its own 1-hour timeout, whichever is
+        // first — see that function's doc comment for why that's fine.
+        {
+            let pipeline_watch = pipeline.clone();
+            std::thread::spawn(move || {
+                match breadcast_core::pipeline::run_until_error_or_timeout(&pipeline_watch, gst::ClockTime::from_seconds(3600))
+                {
+                    Ok(outcome) => tracing::debug!(?outcome, "DLNA encode pipeline bus watcher ended"),
+                    Err(e) => tracing::error!(error = ?e, "DLNA encode pipeline error"),
+                }
+            });
+        }
+
+        pipeline.set_state(gst::State::Playing).context("failed to start the encode pipeline")?;
+
+        let lan_ip = local_lan_ip().context("failed to determine this machine's LAN-reachable IP")?;
+        // Bind an ephemeral port (`:0`) rather than a fixed one like the
+        // `dlna_mirror_test` example uses -- the daemon may need to run
+        // alongside that example, or a future concurrent-session mode,
+        // without a bind conflict. `HttpServer::start`'s worker threads
+        // outlive this session once it stops (documented pre-existing
+        // limitation, see `http_server.rs` -- not something introduced
+        // here); one leaked idle listener per DLNA cast is an accepted
+        // cost until that gets a real shutdown path.
+        let http = breadcast_core::http_server::HttpServer::start("0.0.0.0:0", output_dir.clone())
+            .context("failed to start the HLS HTTP server")?;
+        let stream_url = http.url(lan_ip, "playlist.m3u8");
+
+        wait_for_playlist_segments(&output_dir.join("playlist.m3u8"), 3, Duration::from_secs(20))
+            .await
+            .context("encode pipeline never produced playable HLS segments")?;
+
+        let session = DlnaSession::connect(&device).await.context("failed to connect to the DLNA renderer")?;
+        session.load(&stream_url).await.context("renderer rejected the stream load")?;
+        tracing::info!(device = %device.friendly_name, %stream_url, "DLNA mirroring started");
+
+        let poll_task = {
+            let session = session.clone();
+            let device_name = device.friendly_name.clone();
+            tokio::spawn(async move {
+                loop {
+                    tokio::time::sleep(POLL_INTERVAL).await;
+                    match session.transport_state().await {
+                        Ok(state) if state == "STOPPED" || state == "NO_MEDIA_PRESENT" => {
+                            tracing::info!(device = %device_name, %state, "DLNA renderer ended playback on its own");
+                            let _ = daemon_tx.send(DaemonCommand::SessionEnded).await;
+                            return;
+                        }
+                        Ok(_) => {}
+                        Err(e) => {
+                            tracing::warn!(device = %device_name, error = ?e, "DLNA transport state poll failed, treating renderer as gone");
+                            let _ = daemon_tx.send(DaemonCommand::SessionEnded).await;
+                            return;
+                        }
+                    }
+                }
+            })
+        };
+
+        Ok(Self { pipeline, session, capture: Some(capture), poll_task })
+    }
+
+    /// Tears down the session: stops polling, tells the renderer to stop,
+    /// stops the encode pipeline, and closes the portal capture session.
+    pub async fn stop(mut self) {
+        // A request, not a wait -- if the poll task is mid-poll and sends
+        // one more `SessionEnded` right as this races it, that's harmless:
+        // `daemon.rs`'s handler already no-ops when `active_session` was
+        // already cleared by this explicit stop.
+        self.poll_task.abort();
+
+        if let Err(e) = self.session.stop().await {
+            tracing::warn!(error = ?e, "failed to cleanly stop the DLNA session");
+        }
+        if let Err(e) = self.pipeline.set_state(gst::State::Null) {
+            tracing::warn!(error = ?e, "failed to stop the encode pipeline cleanly");
+        }
+        if let Some(capture) = self.capture.take() {
+            if let Err(e) = capture.close().await {
+                tracing::warn!(error = ?e, "failed to cleanly close the portal capture session");
+            }
+        }
+    }
+}

breadcastd/src/ipc.rs

@@ -0,0 +1,175 @@
+//! breadcastd's private control socket: newline-delimited JSON over a
+//! `UnixListener` at `$XDG_RUNTIME_DIR/breadcast/breadcastd.sock`, separate
+//! from breadd's own pub/sub bus (`bread_events.rs`) — that's for external
+//! automation (a Hyprland keybind emitting `bread.command.cast.*`), this is
+//! for the `breadcast` GTK4 popup's actual live state, which needs pushed
+//! updates a best-effort pub/sub bus doesn't fit as naturally.
+//!
+//! The wire message types (`ClientRequest`/`ServerMessage`/`StateInfo`) live
+//! in `breadcast_core::ipc`, shared with the `breadcast` GUI client so the
+//! two never drift out of sync.
+//!
+//! Multiple clients (in practice: zero or one `breadcast` popup instance at
+//! a time, but nothing here assumes that) can connect concurrently; each
+//! gets its own copy of every broadcast event.
+
+use anyhow::{Context, Result};
+use breadcast_core::ipc::{ClientRequest, ServerMessage, socket_path};
+use serde_json::Value;
+use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
+use tokio::net::{UnixListener, UnixStream};
+use tokio::sync::{broadcast, mpsc};
+
+use crate::daemon::DaemonCommand;
+
+/// Removes any stale socket left behind by an unclean shutdown, then binds
+/// and serves forever. `daemon_tx` is how request handling reaches the
+/// daemon's single state-owning actor (see `daemon.rs`); `events_tx`'s
+/// receiver half is (re-)subscribed per connection, so every connection
+/// sees the same `device_list_changed`/`state_changed` events from the
+/// point it connects onward.
+pub async fn serve(daemon_tx: mpsc::Sender<DaemonCommand>, events_tx: broadcast::Sender<ServerMessage>) -> Result<()> {
+    let socket_path = socket_path()?;
+    if let Some(parent) = socket_path.parent() {
+        std::fs::create_dir_all(parent)
+            .with_context(|| format!("failed to create socket dir {}", parent.display()))?;
+    }
+    // A stale socket file from an unclean previous exit makes bind() fail
+    // with AddrInUse even though nothing is actually listening -- the
+    // daemon's own singleton lock (see main.rs) already guarantees at most
+    // one live breadcastd, so it's always safe to clear this before binding.
+    let _ = std::fs::remove_file(&socket_path);
+
+    let listener = UnixListener::bind(&socket_path)
+        .with_context(|| format!("failed to bind {}", socket_path.display()))?;
+    tracing::info!(path = %socket_path.display(), "IPC socket listening");
+
+    loop {
+        let (stream, _addr) = listener.accept().await.context("failed to accept IPC connection")?;
+        let daemon_tx = daemon_tx.clone();
+        let events_rx = events_tx.subscribe();
+        tokio::spawn(async move {
+            if let Err(e) = handle_connection(stream, daemon_tx, events_rx).await {
+                tracing::debug!(error = %e, "IPC connection ended");
+            }
+        });
+    }
+}
+
+async fn handle_connection(
+    stream: UnixStream,
+    daemon_tx: mpsc::Sender<DaemonCommand>,
+    mut events_rx: broadcast::Receiver<ServerMessage>,
+) -> Result<()> {
+    let (read_half, write_half) = stream.into_split();
+    let mut reader = BufReader::new(read_half).lines();
+
+    // A single task owns the write half and serializes everything written
+    // to it (responses and pushed events both funnel through `writer_tx`)
+    // -- two independent tasks (one per direction) writing directly to the
+    // same stream could interleave partial JSON lines.
+    let (writer_tx, mut writer_rx) = mpsc::unbounded_channel::<ServerMessage>();
+    let writer_task = tokio::spawn(async move {
+        let mut write_half = write_half;
+        while let Some(msg) = writer_rx.recv().await {
+            let Ok(mut line) = serde_json::to_string(&msg) else { continue };
+            line.push('\n');
+            if write_half.write_all(line.as_bytes()).await.is_err() {
+                return;
+            }
+        }
+    });
+
+    let forward_tx = writer_tx.clone();
+    let forward_task = tokio::spawn(async move {
+        loop {
+            match events_rx.recv().await {
+                Ok(event) => {
+                    if forward_tx.send(event).is_err() {
+                        return;
+                    }
+                }
+                Err(broadcast::error::RecvError::Lagged(_)) => continue,
+                Err(broadcast::error::RecvError::Closed) => return,
+            }
+        }
+    });
+
+    while let Some(line) = reader.next_line().await? {
+        if line.trim().is_empty() {
+            continue;
+        }
+        let request: ClientRequest = match serde_json::from_str(&line) {
+            Ok(req) => req,
+            Err(e) => {
+                tracing::debug!(error = %e, %line, "malformed IPC request, ignoring");
+                continue;
+            }
+        };
+        let response = handle_request(request, &daemon_tx).await;
+        if writer_tx.send(response).is_err() {
+            break;
+        }
+    }
+
+    forward_task.abort();
+    drop(writer_tx);
+    let _ = writer_task.await;
+    Ok(())
+}
+
+async fn handle_request(request: ClientRequest, daemon_tx: &mpsc::Sender<DaemonCommand>) -> ServerMessage {
+    let id = request.id;
+    match request.method.as_str() {
+        "list_devices" => {
+            let (reply_tx, reply_rx) = tokio::sync::oneshot::channel();
+            if daemon_tx.send(DaemonCommand::ListDevices { reply: reply_tx }).await.is_err() {
+                return ServerMessage::err(id, "daemon actor has stopped");
+            }
+            match reply_rx.await {
+                Ok(devices) => ServerMessage::ok(id, serde_json::json!(devices)),
+                Err(_) => ServerMessage::err(id, "daemon actor dropped the reply"),
+            }
+        }
+        "get_state" => {
+            let (reply_tx, reply_rx) = tokio::sync::oneshot::channel();
+            if daemon_tx.send(DaemonCommand::GetState { reply: reply_tx }).await.is_err() {
+                return ServerMessage::err(id, "daemon actor has stopped");
+            }
+            match reply_rx.await {
+                Ok(state) => ServerMessage::ok(id, serde_json::json!(state)),
+                Err(_) => ServerMessage::err(id, "daemon actor dropped the reply"),
+            }
+        }
+        "start_cast" => {
+            let Some(device_id) = request.params.get("device_id").and_then(Value::as_str) else {
+                return ServerMessage::err(id, "start_cast requires a string \"device_id\" param");
+            };
+            let (reply_tx, reply_rx) = tokio::sync::oneshot::channel();
+            if daemon_tx
+                .send(DaemonCommand::StartCast { device_id: device_id.to_string(), reply: reply_tx })
+                .await
+                .is_err()
+            {
+                return ServerMessage::err(id, "daemon actor has stopped");
+            }
+            match reply_rx.await {
+                Ok(Ok(())) => ServerMessage::ok(id, Value::Null),
+                Ok(Err(e)) => ServerMessage::err(id, e),
+                Err(_) => ServerMessage::err(id, "daemon actor dropped the reply"),
+            }
+        }
+        "stop_cast" => {
+            let (reply_tx, reply_rx) = tokio::sync::oneshot::channel();
+            if daemon_tx.send(DaemonCommand::StopCast { reply: reply_tx }).await.is_err() {
+                return ServerMessage::err(id, "daemon actor has stopped");
+            }
+            match reply_rx.await {
+                Ok(Ok(())) => ServerMessage::ok(id, Value::Null),
+                Ok(Err(e)) => ServerMessage::err(id, e),
+                Err(_) => ServerMessage::err(id, "daemon actor dropped the reply"),
+            }
+        }
+        other => ServerMessage::err(id, format!("unknown method \"{other}\"")),
+    }
+}

breadcastd/src/main.rs

@@ -0,0 +1,122 @@
+//! breadcastd — background daemon: mDNS (Cast) and SSDP (DLNA) device
+//! discovery, both mirroring session lifecycles (Cast Streaming and
+//! DLNA/AVTransport), and a private IPC socket the `breadcast` GTK4 popup
+//! talks to. Also does optional breadd event integration for external
+//! automation (a Hyprland keybind, etc.) — see `bread_events.rs`; that's
+//! separate from (and does not replace) the IPC socket, since a popup UI
+//! needs live pushed state that a best-effort pub/sub bus doesn't fit as
+//! naturally. See `ipc.rs`'s doc comment.
+
+mod bread_events;
+mod cast_mirror;
+mod daemon;
+mod dlna_mirror;
+mod ipc;
+
+use std::collections::HashMap;
+
+use bread_utils::singleton::{Acquire, try_acquire};
+use breadcast_core::{CastDevice, DlnaDevice, DlnaDiscovery, DlnaDiscoveryEvent, Discovery, DiscoveryEvent};
+use daemon::DaemonCommand;
+
+#[tokio::main]
+async fn main() -> anyhow::Result<()> {
+    tracing_subscriber::fmt::init();
+
+    let _guard = match try_acquire(bread_events::APP_ID)? {
+        Acquire::Acquired(guard) => guard,
+        Acquire::HeldByOther(pid) => {
+            tracing::error!(?pid, "breadcastd already running, exiting");
+            std::process::exit(1);
+        }
+    };
+
+    let bread_client = bread_utils::bread_client::BreadClient::connect(bread_events::APP_ID);
+
+    let (events_tx, _events_rx) = tokio::sync::broadcast::channel(64);
+    let daemon_tx = daemon::spawn(events_tx.clone(), bread_client.clone());
+
+    let commands_daemon_tx = daemon_tx.clone();
+    let _commands = bread_client.subscribe("bread.command.cast.**", move |event| {
+        bread_events::handle_command(&event, &commands_daemon_tx);
+    });
+
+    let ipc_daemon_tx = daemon_tx.clone();
+    tokio::spawn(async move {
+        if let Err(e) = ipc::serve(ipc_daemon_tx, events_tx).await {
+            tracing::error!(error = ?e, "IPC server ended unexpectedly");
+        }
+    });
+
+    let (_discovery, mut cast_events) = Discovery::start()?;
+    let (_dlna_discovery, mut dlna_events) = DlnaDiscovery::start();
+    tracing::info!("breadcastd started, browsing for Cast and DLNA devices");
+
+    // `DiscoveryEvent::Found`/`DlnaDiscoveryEvent::Found` fire on every
+    // re-resolution/re-poll (see `discovery.rs`'s and `dlna/discovery.rs`'s
+    // doc comments), not just the first sighting. Every event is forwarded
+    // to the daemon actor unconditionally (its `HashMap` re-insert is a
+    // harmless no-op for an unchanged device, aside from a redundant but
+    // cheap `device_list_changed` broadcast) — but `bread.cast.device_found`
+    // on the breadd bus is gated on an actual *change*, so anything
+    // subscribed there expecting "on device_found, do X" doesn't fire
+    // repeatedly for the same device.
+    let mut known_cast_devices: HashMap<String, CastDevice> = HashMap::new();
+    let mut known_dlna_devices: HashMap<String, DlnaDevice> = HashMap::new();
+
+    loop {
+        tokio::select! {
+            event = cast_events.recv() => {
+                match event {
+                    Some(DiscoveryEvent::Found(device)) => {
+                        if known_cast_devices.get(&device.id) != Some(&device) {
+                            tracing::info!(?device, "Cast device found");
+                            bread_events::emit_device_found(&bread_client, &device.id, &device.name, &device.model, "cast");
+                            known_cast_devices.insert(device.id.clone(), device.clone());
+                        }
+                        let _ = daemon_tx.send(DaemonCommand::CastDeviceFound(device)).await;
+                    }
+                    Some(DiscoveryEvent::Lost { id }) => {
+                        tracing::info!(%id, "Cast device lost");
+                        known_cast_devices.remove(&id);
+                        let _ = daemon_tx.send(DaemonCommand::CastDeviceLost(id)).await;
+                    }
+                    // The mDNS browse thread itself ended (daemon shutdown,
+                    // an unrecoverable browse error) — that's Cast
+                    // discovery's one job gone, not a normal exit.
+                    // Returning `Ok(())` from `main` here would exit 0, and
+                    // systemd's `Restart=on-failure` would never trigger.
+                    None => {
+                        tracing::error!("Cast discovery channel closed unexpectedly, exiting");
+                        std::process::exit(1);
+                    }
+                }
+            }
+            event = dlna_events.recv() => {
+                match event {
+                    Some(DlnaDiscoveryEvent::Found(device)) => {
+                        if known_dlna_devices.get(&device.url) != Some(&device) {
+                            tracing::info!(?device, "DLNA device found");
+                            bread_events::emit_device_found(&bread_client, &device.url, &device.friendly_name, "DLNA renderer", "dlna");
+                            known_dlna_devices.insert(device.url.clone(), device.clone());
+                        }
+                        let _ = daemon_tx.send(DaemonCommand::DlnaDeviceFound(device)).await;
+                    }
+                    Some(DlnaDiscoveryEvent::Lost { url }) => {
+                        tracing::info!(%url, "DLNA device lost");
+                        known_dlna_devices.remove(&url);
+                        let _ = daemon_tx.send(DaemonCommand::DlnaDeviceLost(url)).await;
+                    }
+                    // Same reasoning as the Cast arm above -- `DlnaDiscovery`'s
+                    // background task only ends via a panic or this process
+                    // dropping every sender, neither of which should happen
+                    // while this loop is still the one holding the receiver.
+                    None => {
+                        tracing::error!("DLNA discovery channel closed unexpectedly, exiting");
+                        std::process::exit(1);
+                    }
+                }
+            }
+        }
+    }
+}