Patch B: drive the .socket2 event stream

Emits openwindow/closewindow/activewindow/workspace by diffing a state
snapshot from lib.rs's existing throttled refresh, rather than instrumenting
window map/unmap and focus-change sites across shell/. One call plus one new
file, so the fork stays rebasable against the ~46 upstream commits a month
those files see.

The snapshot walk is skipped entirely when nothing is subscribed to .socket2,
which is every tick when no bar is running. The baseline is still updated so
a newly-connected client does not receive a burst of openwindow events for
windows that were already open.

Ordering follows Hyprland: openwindow before the activewindow that usually
accompanies it, closewindow after, so a client tracking windows by address
never sees an activewindow for one it has not been told about.

Known trade-off: a window opening and closing inside one 150ms tick produces
no events. Hyprland would emit both.

9 new tests driving a real socketpair through the actual broadcast path.
22 hypr_ipc tests pass.

Runtime-verified nested: subscribing to .socket2 and opening a window yields
  openwindow>>1,1,foot,foot / activewindow>>foot,foot
  activewindow>>, / closewindow>>1

src/hypr_ipc/mod.rs

@@ -24,5 +24,6 @@ pub mod events;
 pub mod protocol;
 pub mod query;
 pub mod socket;
+pub mod sync;
 
 pub use socket::{HyprIpcHandle, init};

src/hypr_ipc/socket.rs

@@ -8,22 +8,18 @@
 //! compositor's own event loop so nothing here spawns a raw thread or
 //! touches compositor state outside of calloop callbacks.
 //!
-//! # Required integration (not applied here — see the executor's final
-//! report for the exact snippets and which files they belong in)
+//! # Integration points
 //!
-//! 1. `src/lib.rs`: add `pub mod hypr_ipc;` and call [`init`] once the
-//!    compositor is ready (same place `session::run_socket` is called).
-//! 2. `src/state.rs`: add a `pub hypr_ipc: crate::hypr_ipc::HyprIpcHandle`
-//!    field to `Common`, populated from the `HyprIpcHandle` [`init`] returns.
-//! 3. `src/session.rs`'s `get_env`: forward `HYPRLAND_INSTANCE_SIGNATURE` to
-//!    spawned children, mirroring how `WAYLAND_DISPLAY`/`DISPLAY` are
-//!    forwarded there.
-//! 4. Optional: call `HyprIpcHandle::notify_*` from the existing
-//!    window/workspace focus-change call sites (e.g.
-//!    `wayland/protocols/toplevel_info.rs`, `shell/mod.rs`) to drive the
-//!    `.socket2` event stream. Without this, request/response (`workspaces`,
-//!    `activewindow`, `clients`, `monitors`) still works; only live event
-//!    notifications are inert until wired up.
+//! 1. `src/lib.rs` calls [`init`] once the compositor is ready, alongside
+//!    `session::run_socket`.
+//! 2. `Common::hypr_ipc` holds the returned handle (`None` until the sockets
+//!    bind, since `Common` is built first).
+//! 3. `src/session.rs`'s `get_env` forwards `HYPRLAND_INSTANCE_SIGNATURE` to
+//!    children, the same way `WAYLAND_DISPLAY` is — that is how clients locate
+//!    these sockets.
+//! 4. The `.socket2` event stream is driven by [`super::sync::refresh`] from
+//!    `lib.rs`'s throttled `refresh`, which diffs a state snapshot rather than
+//!    emitting from scattered call sites.
 
 use std::{
     cell::RefCell,
@@ -58,9 +54,63 @@ impl AsFd for RequestStream {
 #[derive(Clone, Debug)]
 pub struct HyprIpcHandle {
     event_streams: Rc<RefCell<Vec<UnixStream>>>,
+    /// Last state the event stream described. Owned here rather than in
+    /// `Common` so the whole feature stays inside this module.
+    last_snapshot: Rc<RefCell<super::sync::Snapshot>>,
 }
 
 impl HyprIpcHandle {
+    /// Whether anyone is listening on `.socket2`.
+    ///
+    /// With no bar running this is false on every tick, which lets the
+    /// diffing refresh skip walking the window tree entirely.
+    pub fn has_event_subscribers(&self) -> bool {
+        !self.event_streams.borrow().is_empty()
+    }
+
+    pub fn take_snapshot(&self) -> super::sync::Snapshot {
+        self.last_snapshot.borrow().clone()
+    }
+
+    pub fn set_snapshot(&self, snapshot: super::sync::Snapshot) {
+        *self.last_snapshot.borrow_mut() = snapshot;
+    }
+
+    /// A handle wired to a real socket pair, plus a reader for whatever was
+    /// broadcast. Uses the genuine write path rather than a mock so the tests
+    /// cover serialization and the drop-dead-subscriber behaviour too.
+    #[cfg(test)]
+    pub fn for_test() -> (Self, impl Fn() -> Vec<String>) {
+        use std::io::ErrorKind;
+
+        let (tx, rx) = UnixStream::pair().expect("socketpair");
+        rx.set_nonblocking(true).expect("nonblocking");
+
+        let handle = HyprIpcHandle {
+            event_streams: Rc::new(RefCell::new(vec![tx])),
+            last_snapshot: Rc::new(RefCell::new(super::sync::Snapshot::default())),
+        };
+
+        let read_all = move || {
+            let mut buf = Vec::new();
+            let mut chunk = [0u8; 4096];
+            let mut rx = &rx;
+            loop {
+                match rx.read(&mut chunk) {
+                    Ok(0) => break,
+                    Ok(n) => buf.extend_from_slice(&chunk[..n]),
+                    Err(e) if e.kind() == ErrorKind::WouldBlock => break,
+                    Err(_) => break,
+                }
+            }
+            String::from_utf8_lossy(&buf)
+                .split_inclusive('\n')
+                .map(str::to_string)
+                .collect::<Vec<_>>()
+        };
+
+        (handle, read_all)
+    }
     fn broadcast_line(&self, line: &str) {
         let mut streams = self.event_streams.borrow_mut();
         // Best-effort broadcast: a write failure (client gone, buffer full
@@ -262,6 +312,7 @@ pub fn init(handle: &LoopHandle<'static, State>) -> Result<HyprIpcHandle> {
 
     let ipc = HyprIpcHandle {
         event_streams: Rc::new(RefCell::new(Vec::new())),
+        last_snapshot: Rc::new(RefCell::new(super::sync::Snapshot::default())),
     };
 
     register_request_listener(handle, request_listener)?;

src/hypr_ipc/sync.rs

@@ -0,0 +1,235 @@
+// SPDX-License-Identifier: GPL-3.0-only
+
+//! Drives the `.socket2` event stream by diffing compositor state.
+//!
+//! Hyprland emits its events from the call sites that cause them. Doing the
+//! same here would mean instrumenting window map/unmap and every focus-change
+//! path across `shell/`, which is exactly the kind of scattered edit that makes
+//! a fork painful to rebase — upstream runs ~46 commits a month through those
+//! files.
+//!
+//! Instead this snapshots the small amount of state the events actually
+//! describe and diffs it from the existing throttled `refresh` in `lib.rs`
+//! (150 ms, see `LastRefresh`). Nothing new is scheduled, no existing function
+//! is touched, and the whole feature is one call plus this file. A bar polls at
+//! human timescales, so the coalescing a 150 ms tick imposes is not observable
+//! — and it is a genuine improvement over per-call-site emission when a single
+//! action changes several things at once.
+//!
+//! The trade-off worth stating: a window that opens and closes inside one tick
+//! produces no events at all. Hyprland would emit both. Nothing in a status bar
+//! can render a window that never existed for a visible instant, so this is
+//! accepted rather than worked around.
+
+use std::collections::BTreeMap;
+
+use super::query;
+use crate::state::State;
+
+/// The subset of compositor state the event stream describes.
+#[derive(Debug, Default, Clone, PartialEq, Eq)]
+pub struct Snapshot {
+    /// Active workspace name on the focused output.
+    pub active_workspace: Option<String>,
+    /// `(class, title)` of the focused window, if any.
+    pub active_window: Option<(String, String)>,
+    /// address -> `(workspace name, class, title)` for every mapped window.
+    pub windows: BTreeMap<u64, (String, String, String)>,
+}
+
+/// Parse the `0x…` address `ClientInfo` carries back into the integer the
+/// event format needs. Both come from `query::window_address`, so this is
+/// lossless; a malformed value simply drops that window from the diff rather
+/// than emitting an event with a bogus address.
+fn address_of(client_address: &str) -> Option<u64> {
+    u64::from_str_radix(client_address.trim_start_matches("0x"), 16).ok()
+}
+
+/// Build a snapshot from current compositor state.
+pub fn snapshot(state: &State) -> Snapshot {
+    let shell = state.common.shell.read();
+
+    let active_workspace = query::active_workspace(&shell).map(|w| w.name);
+    let active_window =
+        query::active_window(&shell, &state.common).map(|c| (c.class.clone(), c.title.clone()));
+
+    let windows = query::collect_clients(&shell, &state.common)
+        .into_iter()
+        .filter_map(|c| {
+            address_of(&c.address).map(|addr| (addr, (c.workspace.name, c.class, c.title)))
+        })
+        .collect();
+
+    Snapshot {
+        active_workspace,
+        active_window,
+        windows,
+    }
+}
+
+/// Emit the events implied by `previous -> current`.
+///
+/// Ordering matches Hyprland's: `openwindow` before the `activewindow` that
+/// usually accompanies it, and `closewindow` after, so a client that tracks
+/// windows by address never sees an `activewindow` for one it has not been
+/// told about yet.
+pub fn emit_diff(ipc: &super::socket::HyprIpcHandle, previous: &Snapshot, current: &Snapshot) {
+    for (addr, (workspace, class, title)) in &current.windows {
+        if !previous.windows.contains_key(addr) {
+            ipc.notify_open_window(*addr, workspace, class, title);
+        }
+    }
+
+    if previous.active_workspace != current.active_workspace {
+        if let Some(name) = &current.active_workspace {
+            ipc.notify_workspace(name);
+        }
+    }
+
+    if previous.active_window != current.active_window {
+        // Hyprland sends `activewindow>>,` with empty fields when focus is
+        // lost, rather than sending nothing, so bars can clear their label.
+        let (class, title) = current
+            .active_window
+            .clone()
+            .unwrap_or_else(|| (String::new(), String::new()));
+        ipc.notify_active_window(&class, &title);
+    }
+
+    for addr in previous.windows.keys() {
+        if !current.windows.contains_key(addr) {
+            ipc.notify_close_window(*addr);
+        }
+    }
+}
+
+/// Snapshot, diff against the previous tick, emit, store. Called from
+/// `lib.rs`'s `refresh`.
+pub fn refresh(state: &mut State) {
+    // Cloning the handle first releases the borrow on `common` that
+    // `snapshot` needs to take on `common.shell`.
+    let Some(ipc) = state.common.hypr_ipc.clone() else {
+        return;
+    };
+
+    // Skip the walk entirely when nobody is listening — with no bar running
+    // this is every single tick.
+    if !ipc.has_event_subscribers() {
+        // Keep the baseline current so the first subscriber does not receive a
+        // burst of `openwindow` events for windows that were already open.
+        ipc.set_snapshot(snapshot(state));
+        return;
+    }
+
+    let current = snapshot(state);
+    let previous = ipc.take_snapshot();
+    if current != previous {
+        emit_diff(&ipc, &previous, &current);
+    }
+    ipc.set_snapshot(current);
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn snap(
+        ws: Option<&str>,
+        active: Option<(&str, &str)>,
+        windows: &[(u64, &str, &str, &str)],
+    ) -> Snapshot {
+        Snapshot {
+            active_workspace: ws.map(String::from),
+            active_window: active.map(|(c, t)| (c.to_string(), t.to_string())),
+            windows: windows
+                .iter()
+                .map(|(a, w, c, t)| (*a, (w.to_string(), c.to_string(), t.to_string())))
+                .collect(),
+        }
+    }
+
+    /// `emit_diff` drives a real handle, so the tests capture what a subscriber
+    /// would actually receive over `.socket2`.
+    fn lines_for(previous: &Snapshot, current: &Snapshot) -> Vec<String> {
+        let (ipc, rx) = super::super::socket::HyprIpcHandle::for_test();
+        emit_diff(&ipc, previous, current);
+        rx()
+    }
+
+    #[test]
+    fn no_change_emits_nothing() {
+        let s = snap(Some("1"), Some(("foot", "shell")), &[(1, "1", "foot", "shell")]);
+        assert!(lines_for(&s, &s).is_empty());
+    }
+
+    #[test]
+    fn new_window_emits_openwindow() {
+        let before = snap(Some("1"), None, &[]);
+        let after = snap(Some("1"), None, &[(0x1, "1", "foot", "shell")]);
+        let lines = lines_for(&before, &after);
+        assert_eq!(lines.len(), 1, "{lines:?}");
+        assert!(lines[0].starts_with("openwindow>>"), "{}", lines[0]);
+        assert!(lines[0].contains("foot"), "{}", lines[0]);
+    }
+
+    #[test]
+    fn removed_window_emits_closewindow_with_its_address() {
+        let before = snap(Some("1"), None, &[(0x2a, "1", "foot", "shell")]);
+        let after = snap(Some("1"), None, &[]);
+        let lines = lines_for(&before, &after);
+        assert_eq!(lines, vec!["closewindow>>2a\n"]);
+    }
+
+    #[test]
+    fn workspace_change_emits_workspace() {
+        let before = snap(Some("1"), None, &[]);
+        let after = snap(Some("2"), None, &[]);
+        assert_eq!(lines_for(&before, &after), vec!["workspace>>2\n"]);
+    }
+
+    #[test]
+    fn focus_change_emits_activewindow() {
+        let before = snap(Some("1"), Some(("foot", "a")), &[]);
+        let after = snap(Some("1"), Some(("kitty", "b")), &[]);
+        assert_eq!(lines_for(&before, &after), vec!["activewindow>>kitty,b\n"]);
+    }
+
+    #[test]
+    fn losing_focus_emits_empty_activewindow_so_bars_can_clear() {
+        let before = snap(Some("1"), Some(("foot", "a")), &[]);
+        let after = snap(Some("1"), None, &[]);
+        assert_eq!(lines_for(&before, &after), vec!["activewindow>>,\n"]);
+    }
+
+    #[test]
+    fn openwindow_precedes_activewindow_and_closewindow_follows() {
+        // A client tracking windows by address must never see an activewindow
+        // for one it has not been told about.
+        let before = snap(Some("1"), Some(("old", "x")), &[(0x1, "1", "old", "x")]);
+        let after = snap(Some("1"), Some(("new", "y")), &[(0x2, "1", "new", "y")]);
+        let lines = lines_for(&before, &after);
+        let kinds: Vec<&str> = lines
+            .iter()
+            .map(|l| l.split_once(">>").unwrap().0)
+            .collect();
+        assert_eq!(kinds, vec!["openwindow", "activewindow", "closewindow"]);
+    }
+
+    #[test]
+    fn title_change_alone_emits_activewindow() {
+        // Bars display the active window's title, so a retitle must propagate.
+        let before = snap(Some("1"), Some(("foot", "vim")), &[]);
+        let after = snap(Some("1"), Some(("foot", "vim - file.rs")), &[]);
+        assert_eq!(
+            lines_for(&before, &after),
+            vec!["activewindow>>foot,vim - file.rs\n"]
+        );
+    }
+
+    #[test]
+    fn addresses_round_trip_through_the_client_format() {
+        assert_eq!(address_of("0x2a"), Some(0x2a));
+        assert_eq!(address_of("2a"), Some(0x2a));
+        assert_eq!(address_of("not-an-address"), None);
+    }
+}

src/lib.rs

@@ -369,5 +369,6 @@ fn refresh(state: &mut State) {
     OverlapNotifyState::refresh(state);
     state.common.update_x11_stacking_order();
     KeyboardLayoutState::refresh(state);
+    hypr_ipc::sync::refresh(state);
     state.last_refresh = LastRefresh::At(Instant::now());
 }