Add workspace screen sharing (#64)

Add workspace selection to the screen-sharing picker with pinned capture, privacy filtering and bounded capture resources. Refresh captured tab labels when title or app-ID metadata changes.

scripts/tests/test_workspace_capture_isolated.py

@@ -0,0 +1,241 @@
+"""Exercise workspace sharing on an isolated nested desktop and real PipeWire.
+
+FERESE_TEST_WORKSPACE_CAPTURE=1 python3 scripts/tests/test_workspace_capture_isolated.py
+Set FERESE_TEST_CAPTURE_TRANSPORT=dma to require DMA-BUF (default: shm).
+Only synthetic test windows are captured; the host desktop is never a source.
+"""
+import json
+import os
+from pathlib import Path
+import selectors
+import signal
+import subprocess
+import tempfile
+import time
+import unittest
+
+from session_socket import ipc_environment, ipc_socket
+
+REPO = Path(__file__).resolve().parents[2]
+COMPOSITOR = Path(os.environ.get("FERESE_TEST_BINARY", REPO / "target/debug/ferese"))
+CTL = Path(os.environ.get("FERESE_TEST_CTL", REPO / "target/debug/feresectl"))
+PORTAL = Path(os.environ.get("FERESE_TEST_PORTAL", REPO / "target/debug/xdg-desktop-portal-ferese"))
+
+
+@unittest.skipUnless(os.environ.get("FERESE_TEST_WORKSPACE_CAPTURE") == "1", "requires Wayland and PipeWire")
+class WorkspaceCapture(unittest.TestCase):
+    def test_pinned_workspace_damage_callbacks_privacy_and_resume(self):
+        from PIL import Image
+        children = []
+        with tempfile.TemporaryDirectory(prefix="ferese-workspace-capture-") as temporary:
+            root = Path(temporary)
+            protocols = Path(subprocess.check_output(
+                ["pkg-config", "--variable=pkgdatadir", "wayland-protocols"], text=True).strip())
+            xml = protocols / "stable/xdg-shell/xdg-shell.xml"
+            for kind, name in [("client-header", "xdg-shell-client-protocol.h"), ("private-code", "xdg-shell-protocol.c")]:
+                subprocess.run(["wayland-scanner", kind, str(xml), str(root / name)], check=True)
+            for fixture, name, package, extra in [
+                ("window-capture.c", "client", "wayland-client", [str(root / "xdg-shell-protocol.c")]),
+                ("capture-consumer.c", "consumer", "libpipewire-0.3", []),
+            ]:
+                flags = subprocess.check_output(["pkg-config", "--cflags", "--libs", package], text=True).split()
+                subprocess.run(["cc", "-Wall", "-Wextra", "-Werror", "-I", str(root),
+                                str(REPO / "scripts/tests/fixtures" / fixture), *extra,
+                                "-o", str(root / name), *flags], check=True)
+            runtime = root / "runtime"
+            runtime.mkdir(mode=0o700)
+            config = root / "config/ferese/config.kdl"
+            config.parent.mkdir(parents=True)
+            base_config = ('animations { reduced-motion #true; }\n'
+                           'window-rule app-id="ferese.test.window-capture" floating=#true\n')
+            config.write_text(base_config)
+            host_runtime = Path(os.environ["XDG_RUNTIME_DIR"])
+            display = Path(os.environ["WAYLAND_DISPLAY"])
+            if not display.is_absolute():
+                display = host_runtime / display
+            env = dict(os.environ, XDG_RUNTIME_DIR=str(runtime), WAYLAND_DISPLAY=str(display),
+                       XDG_CONFIG_HOME=str(root / "config"), XDG_STATE_HOME=str(root / "state"),
+                       XDG_CACHE_HOME=str(root / "cache"), FERESE_ENABLE_SCREENCOPY="1",
+                       FERESE_TRACE_PERFORMANCE="1", FERESE_TEST_CAPTURE_MARKER_SIZE="16",
+                       PIPEWIRE_REMOTE=str(host_runtime / "pipewire-0"))
+            for key in ("WAYLAND_SOCKET", "FERESE_SHELL_CONTROL_SOCKET", "DISPLAY", "XAUTHORITY"):
+                env.pop(key, None)
+            transport = os.environ.get("FERESE_TEST_CAPTURE_TRANSPORT", "shm")
+            if transport == "shm":
+                env["FERESE_CAPTURE_FORCE_SHM"] = "1"
+            else:
+                env.pop("FERESE_CAPTURE_FORCE_SHM", None)
+            log_path = root / "session.log"
+            with log_path.open("w") as log:
+                def launch(command, **kwargs):
+                    child = subprocess.Popen(command, env=ipc_environment(env), stderr=log,
+                                             stdout=kwargs.pop("stdout", log), start_new_session=True, **kwargs)
+                    children.append(child)
+                    return child
+
+                def until(predicate, message, timeout=10):
+                    deadline = time.monotonic() + timeout
+                    while time.monotonic() < deadline:
+                        result = predicate()
+                        if result:
+                            return result
+                        time.sleep(.02)
+                    self.fail((message() if callable(message) else message) + "\n" + log_path.read_text())
+
+                def call(*args):
+                    return json.loads(subprocess.check_output([str(CTL), "-j", *args], env=ipc_environment(env), timeout=10))
+
+                def change(client, command):
+                    client.stdin.write(command)
+                    client.stdin.flush()
+
+                try:
+                    compositor = launch([str(COMPOSITOR), "--backend=nested"])
+                    until(lambda: ipc_socket(runtime).exists(), "nested compositor did not start")
+                    env["WAYLAND_DISPLAY"] = str(next(path for path in runtime.glob("wayland-*") if not path.name.endswith(".lock")))
+                    target_client = launch([str(root / "client"), "interactive"], stdin=subprocess.PIPE)
+                    until(lambda: len(call("windows")) == 1, "target did not map")
+                    target = call("windows")[0]
+                    workspace = target["workspace"]
+                    output = call("outputs")[0]
+                    call("workspace", "2")
+                    private_client = launch([str(root / "client"), "cover", "interactive"], stdin=subprocess.PIPE)
+                    until(lambda: len(call("windows")) == 2, "private workspace did not map")
+                    private_workspace = next(window["workspace"] for window in call("windows") if window["id"] != target["id"])
+
+                    def usage(process):
+                        fields = Path(f"/proc/{process.pid}/stat").read_text().rsplit(")", 1)[1].split()
+                        ticks = int(fields[11]) + int(fields[12])
+                        rss = int(fields[21]) * os.sysconf("SC_PAGE_SIZE")
+                        return ticks, rss
+
+                    def idle_cpu(processes, seconds=.6):
+                        before = [usage(process) for process in processes]
+                        start = time.monotonic()
+                        time.sleep(seconds)
+                        elapsed = time.monotonic() - start
+                        after = [usage(process) for process in processes]
+                        return [(100 * (end[0] - begin[0]) / os.sysconf("SC_CLK_TCK") / elapsed, end[1])
+                                for begin, end in zip(before, after)]
+
+                    time.sleep(1.5)
+                    baseline = idle_cpu([compositor], seconds=1.0)[0]
+
+                    def worker():
+                        child = launch([str(PORTAL), "--stream-workspace", str(workspace), "hidden"],
+                                       stdin=subprocess.PIPE, stdout=subprocess.PIPE)
+                        with selectors.DefaultSelector() as selector:
+                            selector.register(child.stdout, selectors.EVENT_READ)
+                            self.assertTrue(selector.select(10), log_path.read_text())
+                        ready = json.loads(child.stdout.readline())
+                        self.assertEqual(ready["logical_size"], [output["width"], output["height"]])
+                        self.assertLessEqual(ready["width"] * ready["height"], 1920 * 1080)
+                        return child, ready
+
+                    stream, ready = worker()
+                    frame_bytes = ready["width"] * ready["height"] * 4
+                    raw = root / "frames.bgrx"
+
+                    def consumer(path, node, frames=1000):
+                        return launch([str(root / "consumer"), str(node), transport, str(frames), str(path)])
+
+                    first_consumer = consumer(raw, ready["node"])
+
+                    def count(path=raw):
+                        return path.stat().st_size // frame_bytes if path.exists() else 0
+
+                    def image(path=raw):
+                        with path.open("rb") as frames:
+                            frames.seek((count(path) - 1) * frame_bytes)
+                            return Image.frombytes("RGB", (ready["width"], ready["height"]), frames.read(frame_bytes), "raw", "BGRX")
+
+                    def pixels(frame):
+                        return getattr(frame, "get_flattened_data", frame.getdata)()
+
+                    until(lambda: count() == 1, "initial workspace frame missing")
+                    scale = ready["width"] / output["width"]
+                    red = (round((target["x"] - output["x"] + target["width"] / 2) * scale),
+                           round((target["y"] - output["y"] + target["height"] / 4) * scale))
+                    self.assertEqual(image().getpixel(red), (255, 0, 0))
+                    self.assertNotIn((0, 255, 0), pixels(image()), "private workspace leaked")
+                    time.sleep(.5)
+                    idle = idle_cpu([compositor, stream], seconds=1.0)
+                    print(json.dumps({"transport": transport, "size": [ready["width"], ready["height"]],
+                                      "baseline_compositor_cpu_percent": round(baseline[0], 2),
+                                      "sharing_compositor_cpu_percent": round(idle[0][0], 2),
+                                      "sharing_portal_cpu_percent": round(idle[1][0], 2),
+                                      "compositor_rss_delta_mib": round((idle[0][1] - baseline[1]) / 1024**2, 2),
+                                      "portal_rss_mib": round(idle[1][1] / 1024**2, 2)}), flush=True)
+                    # Unrelated commits must not compose/copy a new source frame.
+                    for _ in range(8):
+                        change(private_client, b"d")
+                        time.sleep(.04)
+                    self.assertEqual(count(), 1, "another workspace woke this capture")
+                    # A callback-only commit starts an animation on a hidden workspace.
+                    change(target_client, b"a")
+                    marker = (round((target["x"] - output["x"] + target["width"] / 2 + 8) * scale),
+                              round((target["y"] - output["y"] + target["height"] / 2 + 8) * scale))
+                    # Capture coalesces intermediate commits under backpressure.
+                    # Require the final animation state, not one frame per commit.
+                    until(lambda: count() >= 2 and image().getpixel(marker) == (255, 255, 3),
+                          lambda: f"hidden animation stalled at {count()} frames, pixel {image().getpixel(marker)}")
+                    animation_count = count()
+                    time.sleep(.2)
+                    self.assertEqual(count(), animation_count, "static workspace emitted duplicates")
+                    # Independent consumers share the composed view and both update.
+                    other_stream, other_ready = worker()
+                    second_raw = root / "second.bgrx"
+                    other_consumer = consumer(second_raw, other_ready["node"])
+                    until(lambda: count(second_raw) == 1, "second consumer did not receive cached frame")
+                    self.assertEqual(image().tobytes(), image(second_raw).tobytes())
+                    for index in range(6):
+                        change(target_client, b"d")
+                        until(lambda: count() >= animation_count + 1 + index and count(second_raw) >= 2 + index,
+                              "one consumer missed source damage")
+                    # A changed privacy rule must clear retained pixels in both views.
+                    previous = count()
+                    config.write_text(base_config + 'window-rule title="Capture target" block-out-from-screencasts=#true\n')
+                    call("reload-config")
+                    until(lambda: count() > previous, "privacy change did not invalidate retained image")
+                    self.assertNotIn((255, 0, 0), pixels(image()))
+                    self.assertNotIn((0, 0, 255), pixels(image()))
+                    previous = count()
+                    config.write_text(base_config)
+                    call("reload-config")
+                    until(lambda: count() > previous, "privacy opt-in did not repaint")
+                    self.assertEqual(image().getpixel(red), (255, 0, 0))
+                    # Pause long enough to release targets, then resume unchanged content.
+                    for child in [first_consumer, other_consumer]:
+                        os.killpg(child.pid, signal.SIGTERM)
+                        child.wait(timeout=5)
+                    time.sleep(2.3)
+                    self.assertIn("released idle workspace capture target", log_path.read_text())
+                    resumed = root / "resumed.bgrx"
+                    resumed_consumer = consumer(resumed, ready["node"], frames=1)
+                    until(lambda: count(resumed) == 1, "resumed consumer did not get an initial frame")
+                    resumed_consumer.wait(timeout=5)
+                    self.assertEqual(resumed_consumer.returncode, 0)
+                    self.assertEqual(image(resumed).getpixel(red), (255, 0, 0))
+                    self.assertEqual(next(ws["id"] for ws in call("workspaces") if ws["focused"]), private_workspace)
+                    # Closing the source's last window removes that inactive workspace.
+                    os.killpg(target_client.pid, signal.SIGTERM)
+                    target_client.wait(timeout=5)
+                    self.assertEqual(stream.wait(timeout=7), 0)
+                    self.assertEqual(other_stream.wait(timeout=7), 0)
+                    self.assertIsNone(compositor.poll(), log_path.read_text())
+                finally:
+                    for child in reversed(children):
+                        if child.poll() is None:
+                            os.killpg(child.pid, signal.SIGTERM)
+                            try:
+                                child.wait(timeout=5)
+                            except subprocess.TimeoutExpired:
+                                os.killpg(child.pid, signal.SIGKILL)
+                                child.wait()
+                        for pipe in [child.stdin, child.stdout]:
+                            if pipe is not None and pipe is not log:
+                                pipe.close()
+
+
+if __name__ == "__main__":
+    unittest.main()