feat(vdisplay): mirror a pinned physical monitor on KWin

P2 of design/per-monitor-portal-capture.md. zkde_screencast's
stream_output records an output KWin already has — the connector name IS
the selection, so there is no dialog, no portal and no chooser for a
background service to answer.

It arrives as a VirtualDisplay reporting DisplayOwnership::External, which
already means "someone else's display, merely mirrored: no keep-alive, no
topology, no reuse". So the rule that we must never disable, move or
"restore" a monitor the user is sitting in front of holds by construction
rather than by everyone remembering it — and create() ignores the client's
requested mode for the same reason: a panel runs at the mode its owner set.

Routing lives in vdisplay::open, the one place every session opens a
display, so the pin cannot be honored on one plane and ignored on another.
The host sets the libei anchor from the same pin (pf-vdisplay must not
depend on pf-inject), which is what makes absolute input land on the
mirrored head instead of a same-sized neighbour.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

crates/pf-vdisplay/src/lib.rs

@@ -289,7 +289,19 @@ pub fn rebuild_probe_active() -> bool {
 }
 
 /// Open the virtual-display driver for `compositor`.
+///
+/// A `PUNKTFUNK_CAPTURE_MONITOR` pin routes to the **mirror** backend instead: the host streams
+/// that physical head and creates no virtual display at all. Deliberately resolved here, at the one
+/// place every session opens a display, so the pin can't be honored on one plane and ignored on
+/// another — it is a host-wide setting (`design/per-monitor-portal-capture.md` §5.3).
 pub fn open(compositor: Compositor) -> Result<Box<dyn VirtualDisplay>> {
+    #[cfg(target_os = "linux")]
+    if let Some(connector) = pf_host_config::config().capture_monitor.as_deref() {
+        return Ok(Box::new(mirror::MirrorDisplay::new(
+            compositor,
+            connector.to_string(),
+        )?));
+    }
     #[cfg(target_os = "linux")]
     {
         match compositor {
@@ -366,6 +378,14 @@ pub mod policy;
 #[path = "vdisplay/monitors.rs"]
 pub mod monitors;
 
+// The monitor-mirror backend: stream a head the compositor ALREADY has (the
+// `PUNKTFUNK_CAPTURE_MONITOR` pin) instead of creating one. Implements `VirtualDisplay` so the
+// session machinery is unchanged, but reports `DisplayOwnership::External` so none of the
+// virtual-display lifecycle policy is applied to someone else's monitor.
+#[cfg(target_os = "linux")]
+#[path = "vdisplay/mirror.rs"]
+mod mirror;
+
 // The pure per-display lifecycle state machine (refcount + linger + pin), platform-neutral and
 // property-tested; the registry executes the side effects its transitions dictate.
 #[path = "vdisplay/lifecycle.rs"]

crates/pf-vdisplay/src/vdisplay/linux/kwin.rs

@@ -32,6 +32,7 @@ use std::sync::mpsc::Sender;
 use std::sync::Arc;
 use std::thread;
 use std::time::Duration;
+use wayland_client::protocol::wl_output::{self, WlOutput};
 use wayland_client::protocol::wl_registry::{self, WlRegistry};
 use wayland_client::{Connection, Dispatch, Proxy, QueueHandle};
 
@@ -914,6 +915,10 @@ struct State {
     node_id: Option<u32>,
     failed: Option<String>,
     closed: bool,
+    /// Every `wl_output` KWin advertises, keyed by the proxy, with its connector name once the
+    /// `name` event arrives. Only the monitor-mirror path ([`stream_existing_output`]) needs these
+    /// — `stream_output` takes a `wl_output` object, so the connector has to be resolved to one.
+    outputs: Vec<(WlOutput, Option<String>)>,
 }
 
 impl Dispatch<WlRegistry, ()> for State {
@@ -934,6 +939,34 @@ impl Dispatch<WlRegistry, ()> for State {
             if interface == Screencast::interface().name {
                 let v = version.min(MAX_VERSION);
                 state.screencast = Some(registry.bind::<Screencast, _, _>(name, v, qh, ()));
+            } else if interface == WlOutput::interface().name {
+                // v4 is where `wl_output.name` (the connector) arrives; bind at least that when the
+                // compositor offers it, else bind what it has and let the resolve fail loudly
+                // rather than mirroring an unidentifiable head.
+                let v = version.min(WL_OUTPUT_MAX_VERSION);
+                let out = registry.bind::<WlOutput, _, _>(name, v, qh, ());
+                state.outputs.push((out, None));
+            }
+        }
+    }
+}
+
+/// `wl_output` version we bind at: 4 brings the `name` event carrying the connector
+/// (`DP-1`, …) — the only way to tell KWin's outputs apart on this connection.
+const WL_OUTPUT_MAX_VERSION: u32 = 4;
+
+impl Dispatch<WlOutput, ()> for State {
+    fn event(
+        state: &mut Self,
+        output: &WlOutput,
+        event: wl_output::Event,
+        _: &(),
+        _: &Connection,
+        _: &QueueHandle<Self>,
+    ) {
+        if let wl_output::Event::Name { name } = event {
+            if let Some(slot) = state.outputs.iter_mut().find(|(o, _)| o == output) {
+                slot.1 = Some(name);
             }
         }
     }
@@ -988,6 +1021,50 @@ fn virtual_output_thread(
     }
 }
 
+/// Start recording the existing KWin output named `connector` (the monitor-mirror path), returning
+/// its PipeWire node id and the keepalive whose drop stops the recording.
+///
+/// `hw_cursor` selects the pointer mode exactly as the virtual-output path does: metadata for a
+/// cursor-channel session, embedded otherwise, so the cursor behaves the same whichever source a
+/// session is on.
+pub(crate) fn stream_existing_output(
+    connector: &str,
+    hw_cursor: bool,
+) -> Result<(u32, Box<dyn Send>)> {
+    let pointer_mode = if hw_cursor {
+        POINTER_METADATA
+    } else {
+        POINTER_EMBEDDED
+    };
+    let (setup_tx, setup_rx) = std::sync::mpsc::channel::<Result<u32, String>>();
+    let stop = Arc::new(AtomicBool::new(false));
+    let stop_thread = stop.clone();
+    let connector_thread = connector.to_string();
+    thread::Builder::new()
+        .name("punktfunk-kwin-mirror".into())
+        .spawn(move || {
+            if let Err(e) = run_existing(&connector_thread, pointer_mode, &setup_tx, &stop_thread) {
+                let _ = setup_tx.send(Err(format!("{e:#}")));
+            }
+        })
+        .context("spawn KWin monitor-mirror thread")?;
+    let node_id = match setup_rx.recv_timeout(Duration::from_secs(20)) {
+        Ok(Ok(v)) => v,
+        Ok(Err(e)) => bail!("KWin monitor mirror failed: {e}"),
+        Err(_) => bail!("timed out recording the KWin output {connector:?}"),
+    };
+    Ok((node_id, Box::new(StopOnDrop(stop)) as Box<dyn Send>))
+}
+
+/// Stops the mirror thread (and thus the recording) when the capturer drops it.
+struct StopOnDrop(Arc<AtomicBool>);
+
+impl Drop for StopOnDrop {
+    fn drop(&mut self) {
+        self.0.store(true, Ordering::Relaxed);
+    }
+}
+
 /// Readiness probe: connect to the KWin Wayland socket, roundtrip the registry, and confirm
 /// the privileged `zkde_screencast` global is actually advertised. This is exactly what
 /// [`run`] needs before it can create a virtual output, so a session-bringup script can poll
@@ -1019,6 +1096,106 @@ pub fn is_available() -> bool {
     probe().is_ok()
 }
 
+/// Stream an **existing** KWin output — the monitor-mirror path
+/// (`design/per-monitor-portal-capture.md` L1). Same privileged global and the same thread/keepalive
+/// shape as the virtual-output path; `stream_output` simply takes a `wl_output` instead of minting
+/// one, so there is no dialog, no portal, and no chooser: the connector name IS the selection.
+///
+/// Returns the PipeWire node id. The thread parks until `stop`, holding the Wayland connection that
+/// is the cast's lifetime — dropping it stops the recording and leaves the monitor untouched (we
+/// never created it, so there is nothing to tear down; §7.1).
+fn run_existing(
+    connector: &str,
+    pointer_mode: u32,
+    setup_tx: &Sender<Result<u32, String>>,
+    stop: &AtomicBool,
+) -> Result<()> {
+    let conn = Connection::connect_to_env()
+        .context("connect to KWin Wayland (is WAYLAND_DISPLAY set to the KWin socket?)")?;
+    let mut queue = conn.new_event_queue();
+    let qh = queue.handle();
+    let _registry = conn.display().get_registry(&qh, ());
+
+    let mut state = State::default();
+    // Two roundtrips: the first processes the globals (binding screencast + every wl_output), the
+    // second drains each output's property burst — the `name` event we resolve the connector by.
+    queue.roundtrip(&mut state).context("registry roundtrip")?;
+    queue
+        .roundtrip(&mut state)
+        .context("wl_output property roundtrip")?;
+
+    let screencast = state.screencast.clone().ok_or_else(|| {
+        anyhow!(
+            "KWin does not expose zkde_screencast_unstable_v1 to this client — install the host's \
+             .desktop (io.unom.Punktfunk.Host.desktop, X-KDE-Wayland-Interfaces) and re-login so \
+             KWin authorizes it, or run KWin with KWIN_WAYLAND_NO_PERMISSION_CHECKS=1 (headless test)"
+        )
+    })?;
+
+    // Resolve the connector to a bound wl_output. A miss is a hard error naming what IS there:
+    // mirroring some other monitor because the requested one is unplugged shows the operator a
+    // screen they did not ask for, which is worse than a session that refuses with a reason.
+    let named: Vec<&str> = state
+        .outputs
+        .iter()
+        .filter_map(|(_, n)| n.as_deref())
+        .collect();
+    let output = state
+        .outputs
+        .iter()
+        .find(|(_, n)| n.as_deref() == Some(connector))
+        .or_else(|| {
+            state.outputs.iter().find(|(_, n)| {
+                n.as_deref()
+                    .is_some_and(|n| n.eq_ignore_ascii_case(connector))
+            })
+        })
+        .map(|(o, _)| o.clone())
+        .ok_or_else(|| {
+            if named.is_empty() {
+                anyhow!(
+                    "KWin advertised no named wl_output (needs wl_output v4 for the connector \
+                     name) — cannot mirror {connector:?}"
+                )
+            } else {
+                anyhow!(
+                    "KWin has no output named {connector:?} — it has: {}",
+                    named.join(", ")
+                )
+            }
+        })?;
+
+    let stream = screencast.stream_output(&output, pointer_mode, &qh, ());
+    tracing::info!(
+        connector,
+        embedded_pointer = pointer_mode != POINTER_METADATA,
+        "KWin: recording an existing output; awaiting PipeWire node"
+    );
+
+    let node_id = loop {
+        queue
+            .blocking_dispatch(&mut state)
+            .context("wayland dispatch (awaiting created)")?;
+        if let Some(node) = state.node_id {
+            break node;
+        }
+        if let Some(e) = state.failed.take() {
+            bail!("stream_output failed: {e}");
+        }
+        if state.closed {
+            bail!("KWin closed the stream before it was created");
+        }
+    };
+    setup_tx
+        .send(Ok(node_id))
+        .map_err(|_| anyhow!("monitor-mirror opener went away"))?;
+
+    park_until_stopped(&conn, &mut queue, &mut state, stop, connector, node_id)?;
+    stream.close();
+    let _ = conn.flush();
+    Ok(())
+}
+
 fn run(
     width: u32,
     height: u32,
@@ -1080,15 +1257,31 @@ fn run(
         .send(Ok(node_id))
         .map_err(|_| anyhow!("virtual-output opener went away"))?;
 
-    // Keep the connection (and thus the virtual output) alive until told to stop, observing
-    // `closed`. blocking_dispatch can't be interrupted, so poll the connection fd with a short
-    // timeout so `stop` is honored within ~200 ms.
+    park_until_stopped(&conn, &mut queue, &mut state, stop, name, node_id)?;
+
+    // Best-effort clean teardown; dropping the connection also makes KWin reclaim the output.
+    stream.close();
+    let _ = conn.flush();
+    Ok(())
+}
+
+/// Keep the connection (and thus the stream) alive until told to stop, observing `closed`.
+/// `blocking_dispatch` can't be interrupted, so poll the connection fd with a short timeout and
+/// honor `stop` within ~200 ms. Shared by the virtual-output and monitor-mirror paths — for a
+/// virtual output this connection IS the output's lifetime; for a mirror it is only the
+/// recording's, and the monitor itself is untouched either way.
+fn park_until_stopped(
+    conn: &Connection,
+    queue: &mut wayland_client::EventQueue<State>,
+    state: &mut State,
+    stop: &AtomicBool,
+    output: &str,
+    node_id: u32,
+) -> Result<()> {
     while !stop.load(Ordering::Relaxed) {
-        queue
-            .dispatch_pending(&mut state)
-            .context("dispatch_pending")?;
+        queue.dispatch_pending(state).context("dispatch_pending")?;
         if state.closed {
-            tracing::warn!(output = %name, node_id, "KWin closed the virtual-output stream");
+            tracing::warn!(output = %output, node_id, "KWin closed the screencast stream");
             break;
         }
         conn.flush().context("wayland flush")?;
@@ -1110,10 +1303,6 @@ fn run(
             let _ = guard.read();
         } // else: timeout or signal — drop the guard, re-check `stop`
     }
-
-    // Best-effort clean teardown; dropping the connection also makes KWin reclaim the output.
-    stream.close();
-    let _ = conn.flush();
     Ok(())
 }
 

crates/pf-vdisplay/src/vdisplay/mirror.rs

@@ -0,0 +1,200 @@
+//! The monitor-mirror display backend: stream a **physical** head the compositor already has,
+//! instead of creating a virtual one. See `design/per-monitor-portal-capture.md`.
+//!
+//! It implements [`VirtualDisplay`] so it drops into the session machinery unchanged — but it
+//! creates nothing, and that difference is load-bearing:
+//!
+//! * [`DisplayOwnership::External`] ("someone else's display, merely mirrored: no keep-alive, no
+//!   topology, no reuse") is exactly the contract for a head we don't own, so the registry's
+//!   pooling/linger/exclusive machinery passes it through. §7.1 of the design doc — never disable,
+//!   re-position or "restore" a monitor the user is sitting in front of — holds *by construction*
+//!   here rather than by everyone remembering it.
+//! * `create()` ignores the requested [`Mode`]. A panel runs at the mode the user set; the client
+//!   scales. Reconfiguring someone's physical display to match a phone would be the same class of
+//!   rudeness as blanking it (§7.3).
+//!
+//! Selection is a **host-level pin** (`PUNKTFUNK_CAPTURE_MONITOR`), not a per-session choice —
+//! the host-pinned decision of record (§5.3), which is also what makes the input anchor below
+//! sound: one pin for the whole host, so the shared injector can't be re-aimed per session.
+
+use super::backend::{DisplayOwnership, VirtualDisplay, VirtualOutput};
+use super::monitors;
+use crate::{Compositor, Mode};
+use anyhow::{bail, Context, Result};
+
+/// Streams an existing monitor, named by connector.
+pub struct MirrorDisplay {
+    compositor: Compositor,
+    connector: String,
+    hw_cursor: bool,
+}
+
+impl MirrorDisplay {
+    pub fn new(compositor: Compositor, connector: String) -> Result<Self> {
+        Ok(MirrorDisplay {
+            compositor,
+            connector,
+            hw_cursor: false,
+        })
+    }
+}
+
+impl VirtualDisplay for MirrorDisplay {
+    fn name(&self) -> &'static str {
+        "mirror"
+    }
+
+    fn set_hw_cursor(&mut self, on: bool) {
+        self.hw_cursor = on;
+    }
+
+    fn hw_cursor(&self) -> bool {
+        self.hw_cursor
+    }
+
+    fn create(&mut self, _mode: Mode) -> Result<VirtualOutput> {
+        // Resolve the pin against the live head list FIRST: it yields the geometry the input anchor
+        // needs, and it turns "that monitor is gone" into one clear error before any compositor
+        // call. `resolve` refuses to substitute a different head (see its doc).
+        let monitors = monitors::list(self.compositor)
+            .with_context(|| format!("enumerate monitors to mirror {:?}", self.connector))?;
+        let target = monitors::resolve(&monitors, &self.connector)?;
+        check_mirrorable(target)?;
+        let origin = (target.x, target.y);
+        let dims = (target.width, target.height, refresh_hz(target.refresh_mhz));
+
+        let (node_id, keepalive) = match self.compositor {
+            #[cfg(target_os = "linux")]
+            Compositor::Kwin => {
+                crate::kwin::stream_existing_output(&target.connector, self.hw_cursor)?
+            }
+            other => bail!(
+                "mirroring an existing monitor is not implemented for the {} backend yet — \
+                 unset PUNKTFUNK_CAPTURE_MONITOR",
+                other.id()
+            ),
+        };
+
+        // NOTE: aiming absolute input at this head is the HOST's job, not ours — this crate must
+        // not depend on pf-inject (see the crate doc: "never on capture/inject"). The host sets the
+        // anchor from the same pin at startup; §7.2 of the design doc explains why it is host-level
+        // rather than set here per session.
+        tracing::info!(
+            connector = %target.connector,
+            mode = %target.mode_label(),
+            at = %format!("+{}+{}", origin.0, origin.1),
+            node_id,
+            "mirroring an existing monitor (no virtual display created)"
+        );
+
+        // The keepalive IS the recording: dropping it stops the cast and leaves the monitor exactly
+        // as it was (we created nothing, so there is nothing to restore).
+        let mut out = VirtualOutput::owned(node_id, Some(dims), keepalive);
+        // Never pooled, never lingered, never made primary/exclusive: we don't own this head.
+        out.ownership = DisplayOwnership::External;
+        Ok(out)
+    }
+}
+
+/// Can this head be mirrored at all? Both rejections are things a compositor would answer with
+/// silence or a black stream, so they are caught here where the reason can be stated.
+fn check_mirrorable(target: &monitors::PhysicalMonitor) -> Result<()> {
+    if !target.enabled {
+        bail!(
+            "monitor {:?} is disabled — enable it before streaming it",
+            target.connector
+        );
+    }
+    if target.managed {
+        bail!(
+            "monitor {:?} is one of punktfunk's own virtual displays, not a physical head — unset \
+             PUNKTFUNK_CAPTURE_MONITOR to use the normal virtual-display path",
+            target.connector
+        );
+    }
+    if target.width == 0 || target.height == 0 {
+        bail!(
+            "monitor {:?} reports no current mode ({}x{}) — it is not driving a signal",
+            target.connector,
+            target.width,
+            target.height
+        );
+    }
+    Ok(())
+}
+
+/// mHz → whole Hz for [`VirtualOutput::preferred_mode`], never 0 (the negotiation treats 0 as
+/// "unset", and a head that didn't report a refresh is still running at *some* rate — 60 is the
+/// same assumption KWin's own virtual outputs are born with).
+fn refresh_hz(mhz: u32) -> u32 {
+    let hz = (mhz as f64 / 1000.0).round() as u32;
+    if hz == 0 {
+        60
+    } else {
+        hz
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn head(connector: &str) -> monitors::PhysicalMonitor {
+        monitors::PhysicalMonitor {
+            connector: connector.into(),
+            description: "ACME 27".into(),
+            width: 2560,
+            height: 1440,
+            refresh_mhz: 144_000,
+            x: 1920,
+            y: 0,
+            scale: 1.0,
+            primary: false,
+            enabled: true,
+            managed: false,
+        }
+    }
+
+    #[test]
+    fn a_real_enabled_head_is_mirrorable() {
+        assert!(check_mirrorable(&head("DP-2")).is_ok());
+    }
+
+    /// Streaming a dark head would be a black rectangle with no explanation — say why instead.
+    #[test]
+    fn a_disabled_head_is_refused_with_the_reason() {
+        let mut m = head("DP-2");
+        m.enabled = false;
+        let err = check_mirrorable(&m).unwrap_err().to_string();
+        assert!(err.contains("disabled"), "{err}");
+    }
+
+    /// Mirroring our OWN virtual display is a loop with extra steps; catch the misconfiguration
+    /// rather than streaming something confusing.
+    #[test]
+    fn one_of_our_own_virtual_displays_is_refused() {
+        let mut m = head("Virtual-punktfunk-1");
+        m.managed = true;
+        let err = check_mirrorable(&m).unwrap_err().to_string();
+        assert!(err.contains("virtual displays"), "{err}");
+    }
+
+    /// A head listed but not driving a mode (enabled yet modeless) would negotiate a 0x0 stream.
+    #[test]
+    fn a_head_with_no_current_mode_is_refused() {
+        let mut m = head("DP-2");
+        m.width = 0;
+        m.height = 0;
+        let err = check_mirrorable(&m).unwrap_err().to_string();
+        assert!(err.contains("no current mode"), "{err}");
+    }
+
+    #[test]
+    fn refresh_rounds_and_never_reports_zero() {
+        assert_eq!(refresh_hz(60000), 60);
+        assert_eq!(refresh_hz(59940), 60);
+        assert_eq!(refresh_hz(119_920), 120);
+        // An unreported refresh must not become a 0 Hz preferred mode.
+        assert_eq!(refresh_hz(0), 60);
+    }
+}