updated dank to support aqueous

scripts/test-aqueous-integration.py

@@ -0,0 +1,420 @@
+#!/usr/bin/env python3
+"""Run the real DMS daemon/UI on a private Aqueous headless display.
+
+Requires a diagnostic Pixman-compatible Aqueous build and its source checkout.
+All configuration, sockets, images and logs stay in the printed temporary directory.
+"""
+
+import argparse
+import json
+import os
+from pathlib import Path
+import signal
+import struct
+import subprocess
+import tempfile
+import time
+
+
+def wait_for(operation, timeout=15):
+    deadline = time.monotonic() + timeout
+    while time.monotonic() < deadline:
+        value = operation()
+        if value:
+            return value
+        time.sleep(0.1)
+    raise AssertionError("timed out waiting for authoritative state")
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--aqueous-source", type=Path, required=True)
+    parser.add_argument("--bin-dir", type=Path, required=True, help="aqueous, aqueousctl, aqueous-config and dms binaries")
+    parser.add_argument("--frame", action="store_true", help="exercise connected frame reservations on both outputs")
+    parser.add_argument("--xwayland", action="store_true", help="also capture a synthetic XWayland window; uses positive output origins")
+    parser.add_argument("--force-ext", action="store_true", help="verify the actual workspace widget through ext-workspace")
+    args = parser.parse_args()
+    source = args.aqueous_source.resolve() / "compositor"
+    binaries = args.bin_dir.resolve()
+    root = Path(__file__).resolve().parents[1]
+    base = Path(tempfile.mkdtemp(prefix="dms-aqueous-integration-"))
+    print(f"Evidence: {base}", flush=True)
+    for name in ("runtime", "config", "state", "cache", "home", "protocols"):
+        (base / name).mkdir(mode=0o700)
+    env = dict(os.environ)
+    for key in list(env):
+        if key.startswith("AQUEOUS_") or key in ("WAYLAND_DISPLAY", "DISPLAY", "LD_PRELOAD", "DMS_SOCKET"):
+            env.pop(key)
+    env.update(HOME=str(base / "home"), XDG_RUNTIME_DIR=str(base / "runtime"),
+               XDG_CONFIG_HOME=str(base / "config"), XDG_STATE_HOME=str(base / "state"),
+               XDG_CACHE_HOME=str(base / "cache"), PATH=str(binaries) + ":" + env["PATH"],
+               WLR_BACKENDS="headless", WLR_HEADLESS_OUTPUTS="2", WLR_RENDERER="pixman",
+               QT_QPA_PLATFORM="wayland", QT_QUICK_BACKEND="software", GSETTINGS_BACKEND="memory",
+               DMS_FORCE_EXTWS="1" if args.force_ext else "0", DMS_NO_DDC="1", DMS_DISABLE_MATUGEN="1", DMS_DISABLE_HOT_RELOAD="1",
+               DBUS_SESSION_BUS_ADDRESS="unix:path=" + str(base / "missing-session-bus"),
+               DBUS_SYSTEM_BUS_ADDRESS="unix:path=" + str(base / "missing-system-bus"))
+    config = base / "config/aqueous"
+    config.mkdir()
+    shell_config = base / "config/DankMaterialShell"
+    shell_config.mkdir()
+    (shell_config / ".firstlaunch").touch()
+    (shell_config / "settings.json").write_text(json.dumps(dict(
+        frameEnabled=args.frame, frameMode="connected", showDock=True, dockGroupByApp=False,
+        showWorkspaceApps=True, showWorkspacePadding=True, groupWorkspaceApps=False,
+        workspaceFollowFocus=False, showOccupiedWorkspacesOnly=False,
+        barConfigs=[dict(id="default", name="Test", enabled=True, position=0,
+                        screenPreferences=["all"], showOnLastDisplay=True,
+                        leftWidgets=["workspaceSwitcher", "focusedWindow", "runningApps", "keyboard_layout_name"],
+                        centerWidgets=[], rightWidgets=[], spacing=4, innerPadding=4, visible=True)])))
+    for name in ("wm", "layout", "input", "outputs", "rules", "appearance"):
+        path = config / f"{name}.toml"
+        path.write_text((source / "scripts/fixtures/overview-wm.toml").read_text() if name == "wm" else "")
+        env["AQUEOUS_" + ("CONFIG" if name == "wm" else name.upper())] = str(path)
+
+    protocols = {
+        "xdg-shell": Path("/usr/share/wayland-protocols/stable/xdg-shell/xdg-shell.xml"),
+        "shortcuts": Path("/usr/share/wayland-protocols/unstable/keyboard-shortcuts-inhibit/keyboard-shortcuts-inhibit-unstable-v1.xml"),
+        "virtual-keyboard": source / "protocol/upstream/virtual-keyboard-unstable-v1.xml",
+        "layer-shell": source / "protocol/upstream/wlr-layer-shell-unstable-v1.xml",
+        "session-lock": Path("/usr/share/wayland-protocols/staging/ext-session-lock/ext-session-lock-v1.xml"),
+        "aqueous-shell": source / "protocol/aqueous-shell-v1.xml",
+        "ext-workspace": source / "protocol/upstream/ext-workspace-v1.xml",
+    }
+    generated = []
+    for name, xml in protocols.items():
+        subprocess.run(["wayland-scanner", "client-header", str(xml), str(base / "protocols" / f"{name}-client-protocol.h")], check=True)
+        code = base / "protocols" / f"{name}.c"
+        subprocess.run(["wayland-scanner", "private-code", str(xml), str(code)], check=True)
+        generated.append(str(code))
+    fixture = base / "shell-client"
+    subprocess.run(["cc", "-Wall", "-Wextra", "-Werror", "-I" + str(base / "protocols"),
+                    str(source / "scripts/fixtures/shell-client.c"), *generated,
+                    "-lwayland-client", "-lxkbcommon", "-o", str(fixture)], check=True)
+    if args.xwayland:
+        subprocess.run(["cc", "-Wall", "-Wextra", "-Werror", str(source / "scripts/fixtures/shell-x11.c"),
+                        "-lX11", "-o", str(base / "shell-x11")], check=True)
+    children = []
+
+    def spawn(command, log_name):
+        with (base / log_name).open("w") as log:
+            child = subprocess.Popen(command, env=env, stdin=subprocess.PIPE, stdout=log, stderr=log, start_new_session=True)
+        children.append(child)
+        return child
+
+    def run(command, success=True):
+        result = subprocess.run(command, env=env, capture_output=True, text=True, timeout=20)
+        assert (result.returncode == 0) == success, (command, result.stdout, result.stderr)
+        return result.stdout
+
+    def ipc(*args):
+        return run([str(binaries / "dms"), "ipc", "call", *args]).strip()
+
+    def rpc(method, params=None, success=True):
+        assert ipc("aqueousTest", "request", method, json.dumps(params)) == "REQUESTED"
+        def reply():
+            output = ""
+            while True:
+                part = json.loads(ipc("aqueousTest", "result", str(len(output))))
+                output += part["data"]
+                if part["complete"]:
+                    return json.loads(output)
+        response = wait_for(reply)
+        assert ("error" not in response) == success, response
+        return response.get("result") if success else response["error"]
+
+    def state():
+        return json.loads(ipc("aqueousTest", "state"))
+
+    def entities(kind):
+        return [e for e in state().get("batch", {}).get("upsert", []) if e["kind"] == kind]
+
+    def command(action, success=True, **fields):
+        return rpc("aqueous.command", dict(action=action, session=session, **fields), success)
+
+    def watchers():
+        result = []
+        for process in Path("/proc").iterdir():
+            if not process.name.isdigit():
+                continue
+            try:
+                argv = (process / "cmdline").read_bytes().split(b"\0")
+                if argv[1:4] != [b"shell", b"watch", b"--json"]:
+                    continue
+                environment = (process / "environ").read_bytes().split(b"\0")
+            except OSError:
+                continue
+            if argv[1:4] == [b"shell", b"watch", b"--json"] and ("XDG_RUNTIME_DIR=" + str(base / "runtime")).encode() in environment:
+                parent = next(line.split()[1] for line in (process / "status").read_text().splitlines() if line.startswith("PPid:"))
+                assert "dms" not in Path("/proc", parent, "comm").read_text().lower(), "core owns the watcher"
+                result.append(process.name)
+        return result
+
+    # Test-only IPC invokes the production singleton inside the actual shell.
+    shell = base / "shell"
+    shell.mkdir()
+    for path in (root / "quickshell").iterdir():
+        if path.name != "shell.qml":
+            (shell / path.name).symlink_to(path)
+    harness = """
+    property var aqueousTestReply: null
+    property var rememberedWorkspace: null
+    WorkspaceWidgets.WorkspaceSwitcher {
+        id: workspaceTest
+        parentScreen: Quickshell.screens.find(s => s.name === screenName) || null
+        screenName: Quickshell.screens[0]?.name || ""
+        widgetHeight: 30
+        barThickness: 48
+    }
+    function activateWorkspaceIcon(item, windowId) {
+        for (const child of item.children || []) {
+            if (child.windowId === windowId && typeof child.windowAction === "function") {
+                child.windowAction();
+                return true;
+            }
+            if (activateWorkspaceIcon(child, windowId))
+                return true;
+        }
+        return false;
+    }
+    function workspaceDelegates(item) {
+        let result = [];
+        for (const child of item.children || []) {
+            if (child.stableIconCount !== undefined) {
+                result.push({active: child.isActive, placeholder: child.isPlaceholder,
+                    occupied: child.isOccupied, known: child.occupancyKnown, rawWindows: child.workspaceRowData?.windows?.length ?? null, count: child.stableIconCount,
+                    icons: child.workspaceIcons.length, width: child.width, height: child.height,
+                    extraWidth: child.iconsExtraWidth, extraHeight: child.iconsExtraHeight});
+            } else {
+                result = result.concat(workspaceDelegates(child));
+            }
+        }
+        return result;
+    }
+    IpcHandler {
+        target: "aqueousTest"
+        function state(): string { return JSON.stringify(AqueousService.state); }
+        function workspaceStatus(): string {
+            return JSON.stringify({backend: CompositorService.workspaceBackend, output: workspaceTest.effectiveScreenName,
+                rows: workspaceTest.workspaceList.map(w => ({name: w.name, active: w.active, number: w.number,
+                    placeholder: w.placeholder, windows: w.windows?.length ?? null, canActivate: w.canActivate})),
+                delegates: entrypoint.workspaceDelegates(workspaceTest)});
+        }
+        function workspaceOptions(options: string): string {
+            const value = JSON.parse(options);
+            if (value.output !== undefined) workspaceTest.screenName = value.output;
+            if (value.vertical !== undefined) workspaceTest.isVertical = value.vertical;
+            if (value.padding !== undefined) SettingsData.showWorkspacePadding = value.padding;
+            if (value.occupiedOnly !== undefined) SettingsData.showOccupiedWorkspacesOnly = value.occupiedOnly;
+            if (value.group !== undefined) { SettingsData.groupWorkspaceApps = value.group; SettingsData.groupActiveWorkspaceApps = value.group; }
+            if (value.follow !== undefined) SettingsData.workspaceFollowFocus = value.follow;
+            if (value.maxIcons !== undefined) SettingsData.maxWorkspaceIcons = value.maxIcons;
+            return "OK";
+        }
+        function workspaceActivate(index: int): string {
+            workspaceTest.switchToWorkspaceByModelData(workspaceTest.workspaceList[index]);
+            return "REQUESTED";
+        }
+        function workspaceWheel(direction: string): string { workspaceTest.switchWorkspace(direction === "previous" ? -1 : 1); return "REQUESTED"; }
+        function workspaceIcon(windowId: string): string { return String(entrypoint.activateWorkspaceIcon(workspaceTest, windowId)); }
+        function workspaceRemember(): string {
+            entrypoint.rememberedWorkspace = workspaceTest.workspaceList.find(w => !w.active && !w.placeholder);
+            return String(!!entrypoint.rememberedWorkspace);
+        }
+        function workspaceReplay(): string { entrypoint.rememberedWorkspace.activate(); return "REQUESTED"; }
+        function workspaceOverview(): string { return String(CompositorService.toggleWorkspaceOverview(workspaceTest.effectiveScreenName)); }
+        function result(offset: int): string {
+            const text = JSON.stringify(entrypoint.aqueousTestReply);
+            return JSON.stringify({data: text.slice(offset, offset + 8192), complete: offset + 8192 >= text.length});
+        }
+        function request(method: string, params: string): string {
+            entrypoint.aqueousTestReply = null;
+            const fields = JSON.parse(params);
+            if (method === "aqueous.command") {
+                AqueousService.command(fields.action, fields, (ok, status) => {
+                    entrypoint.aqueousTestReply = ok ? {result: {ok: true, status: status}} : {error: status};
+                });
+            } else {
+                AqueousConfigService.request(method.split(".").pop(), fields, (result, error) => {
+                    entrypoint.aqueousTestReply = error ? {error: error} : {result: result};
+                });
+            }
+            return "REQUESTED";
+        }
+    }
+    """
+    source_shell = (root / "quickshell/shell.qml").read_text().replace("import Quickshell\n", "import Quickshell\nimport Quickshell.Io\nimport qs.Modules.DankBar.Widgets as WorkspaceWidgets\n")
+    (shell / "shell.qml").write_text(source_shell.replace("    id: entrypoint", "    id: entrypoint\n" + harness))
+
+    try:
+        compositor = spawn([str(binaries / "aqueous"), *([] if args.xwayland else ["-no-xwayland"]),
+                            "-c", 'printf %s "$DISPLAY" > "$XDG_RUNTIME_DIR/test-display"'], "compositor.log")
+        display = wait_for(lambda: next((p for p in (base / "runtime").glob("wayland-*") if p.is_socket()), None))
+        env["WAYLAND_DISPLAY"] = display.name
+        env["DMS_SHELL_DIR"] = str(shell)
+        dms = spawn([str(binaries / "dms"), "-c", str(shell), "run"], "dms.log")
+        wait_for(lambda: subprocess.run([str(binaries / "dms"), "ipc", "call", "aqueous", "status"], env=env, capture_output=True, text=True, timeout=5).returncode == 0)
+        status = wait_for(lambda: (s if (s := json.loads(ipc("aqueous", "status")))["available"] else None))
+        assert status["outputs"] == 2, status
+        session = status["session"]
+        def workspace_status():
+            value = json.loads(ipc("aqueousTest", "workspaceStatus"))
+            (base / "workspace-status.json").write_text(json.dumps(value, indent=2))
+            return value
+        def workspace_options(**options):
+            assert ipc("aqueousTest", "workspaceOptions", json.dumps(options)) == "OK"
+        view = wait_for(lambda: (v if (v := workspace_status())["delegates"] else None))
+        assert view["backend"] == ("ext" if args.force_ext else "aqueous"), view
+        assert len(view["delegates"]) == len(view["rows"]), view
+        if args.frame:
+            wait_for(lambda: len(entities("output")) == 2 and all(o["usable_bounds"]["width"] < o["bounds"]["width"] and o["usable_bounds"]["height"] < o["bounds"]["height"] for o in entities("output")))
+        wait_for(lambda: ipc("settings", "openWith", "typography").startswith("SETTINGS_OPEN_SUCCESS"))
+        time.sleep(2)
+        assert "Type SettingsModal unavailable" not in (base / "dms.log").read_text()
+        assert ipc("settings", "close") == "SETTINGS_CLOSE_SUCCESS"
+        windows = [spawn([str(fixture), "window"], f"window-{i}.log") for i in range(2)]
+        clients = wait_for(lambda: (w if len(w := [w for w in entities("window") if w["app_id"] == "aq-shell-test"]) == 2 else None))
+        assert clients[0]["title"] == clients[1]["title"] and clients[0]["id"] != clients[1]["id"]
+        seat = entities("seat")[0]["id"]
+        window = clients[0]["id"]
+        assert command("window.activate", id=window, seat=seat)["status"] == "applied"
+        wait_for(lambda: next(s for s in entities("seat") if s["id"] == seat)["window"] == window)
+        group = next(s for s in entities("seat") if s["id"] == seat)["keyboard"]
+        assert command("keyboard.set", seat=seat, group=group, index=1)["status"] == "applied"
+        wait_for(lambda: next(k for k in entities("keyboard") if k["id"] == group)["index"] == 1)
+        assert "German" in json.loads(ipc("aqueous", "status"))["keyboardLayout"]
+        output = next(o for o in entities("output") if o["id"] == next(w for w in entities("window") if w["id"] == window)["output"])
+        workspace_options(output=output["name"], group=False, occupiedOnly=not args.force_ext)
+        if not args.force_ext:
+            view = wait_for(lambda: (v if (v := workspace_status()) and any(d["icons"] == 2 for d in v["delegates"]) else None))
+            assert all(d["icons"] == d["count"] for d in view["delegates"]), view
+            assert len(view["rows"]) == 3 and sum(r["placeholder"] for r in view["rows"]) == 2, view
+            assert any(d["extraWidth"] > 0 for d in view["delegates"]), view
+            workspace_options(group=True)
+            wait_for(lambda: any(d["icons"] == 1 for d in workspace_status()["delegates"]))
+            workspace_options(group=False, vertical=True, maxIcons=1)
+            view = wait_for(lambda: (v if (v := workspace_status()) and any(d["icons"] == 2 and d["extraHeight"] > 0 for d in v["delegates"]) else None))
+            assert all(d["icons"] == d["count"] for d in view["delegates"]), view
+            workspace_options(padding=False)
+            wait_for(lambda: len(workspace_status()["rows"]) == 1)
+            workspace_options(padding=True)
+            wait_for(lambda: len(workspace_status()["rows"]) == 3)
+        else:
+            view = workspace_status()
+            assert all(row["windows"] is None for row in view["rows"]), view
+            assert all(d["icons"] == d["count"] == 0 and not d["known"] for d in view["delegates"]), view
+        workspace_options(vertical=False, maxIcons=3, occupiedOnly=False)
+        view = workspace_status()
+        current = next(i for i, r in enumerate(view["rows"]) if r["active"])
+        target = next(i for i, r in enumerate(view["rows"]) if not r["active"] and not r["placeholder"])
+        assert ipc("aqueousTest", "workspaceActivate", str(target)) == "REQUESTED"
+        wait_for(lambda: workspace_status()["rows"][target]["active"])
+        assert ipc("aqueousTest", "workspaceWheel", "previous" if target > current else "next") == "REQUESTED"
+        wait_for(lambda: workspace_status()["rows"][current]["active"])
+        # Names change independently of identity, including collisions in one projection.
+        same_output = [w for w in entities("workspace") if w["output"] == output["id"]]
+        for workspace in same_output[:2]:
+            command("workspace.rename", id=workspace["id"], name="Duplicate")
+        wait_for(lambda: sum(r["name"] == "Duplicate" for r in workspace_status()["rows"]) == 2)
+        command("workspace.rename", id=same_output[0]["id"], name="Renamed")
+        wait_for(lambda: any(r["name"] == "Renamed" for r in workspace_status()["rows"]))
+        other_output = next(o for o in entities("output") if o["id"] != output["id"])
+        workspace_options(output=other_output["name"], follow=True)
+        wait_for(lambda: workspace_status()["output"] == output["name"])
+        workspace_options(follow=False)
+        assert workspace_status()["output"] == other_output["name"]
+        workspace_options(output=output["name"])
+        if not args.force_ext:
+            command("window.activate", id=window, seat=seat)
+            wait_for(lambda: ipc("aqueousTest", "workspaceIcon", clients[1]["id"]) == "true")
+            wait_for(lambda: next(s for s in entities("seat") if s["id"] == seat)["window"] == clients[1]["id"])
+            command("window.activate", id=window, seat=seat)
+        assert ipc("aqueousTest", "workspaceOverview") == "true"
+        wait_for(lambda: entities("session")[0]["overview_output"] == output["id"])
+        command("overview.hide")
+        assert command("overview.show", output=output["id"])["status"] == "applied"
+        wait_for(lambda: entities("session")[0]["overview_output"] == output["id"])
+        assert ipc("aqueous", "overview", "hide", output["name"]) == "OVERVIEW_REQUESTED"
+        wait_for(lambda: entities("session")[0]["overview_output"] is None)
+
+        def capture(name):
+            command("window.activate", id=window, seat=seat)
+            run([str(binaries / "dms"), "screenshot", "window", "--seat", seat, "--no-clipboard", "--no-notify", "--dir", str(base), "--filename", name])
+            data = (base / name).read_bytes()
+            assert data.startswith(b"\x89PNG\r\n\x1a\n")
+            dimensions = struct.unpack(">II", data[16:24])
+            assert all(dimensions), dimensions
+            print(f"Screenshot {name}: {dimensions}", flush=True)
+
+        capture("window.png")
+        run([str(binaries / "dms"), "screenshot", "full", "--seat", seat, "--geometry"])
+        origin = "3000,100" if args.xwayland else "-3000,-100"
+        run(["wlr-randr", "--output", output["name"], "--scale", "1.25", "--transform", "90", "--pos=" + origin])
+        wait_for(lambda: next(o for o in entities("output") if o["id"] == output["id"])["scale"] == 1.25)
+        capture("rotated-fractional.png" if args.xwayland else "rotated-fractional-negative-origin.png")
+        if args.xwayland:
+            env["DISPLAY"] = wait_for(lambda: (base / "runtime/test-display").read_text())
+            spawn([str(base / "shell-x11")], "xwayland-window.log")
+            xwindow = wait_for(lambda: next((w for w in entities("window") if w["backend"] == "xwayland"), None))
+            window = xwindow["id"]
+            capture("xwayland-window.png")
+            assert command("window.close", id=window)["status"] == "accepted"
+
+        snapshot = rpc("aqueous.config.snapshot")
+        (base / "helper-snapshot.json").write_text(json.dumps(snapshot, indent=2) + "\n")
+        (base / "helper-version.json").write_text(run([str(binaries / "aqueous-config"), "version"]))
+        generation = snapshot["generation"]
+        edit = dict(expected_generation=generation, changes=[dict(id="toggle_overview", value=["Super+W", "Super+F12"])])
+        rpc("aqueous.config.validate", edit)
+        assert rpc("aqueous.config.snapshot")["generation"] == generation
+        saved = rpc("aqueous.config.apply", edit)
+        assert saved["generation"] != generation
+        assert "external_change" in rpc("aqueous.config.apply", edit, success=False)
+        sheet = json.loads(run([str(binaries / "dms"), "keybinds", "show", "aqueous"]))
+        assert sheet["generation"] == saved["generation"]
+        run([str(binaries / "dms"), "keybinds", "set", "aqueous", "Super+F11", "spawn true", "--expected-generation", saved["generation"]])
+        assert "not_found" in command("window.activate", id="removed-window", seat=seat, success=False)
+        assert "stale session" in rpc("aqueous.command", dict(action="window.close", session="0" * 32, id=window), success=False)
+
+        watch_pids = watchers()
+        assert len(watch_pids) == 1, watch_pids
+        assert ipc("aqueousTest", "workspaceRemember") == "true"
+        os.kill(int(watch_pids[0]), signal.SIGKILL)
+        wait_for(lambda: not state().get("available", False))
+        wait_for(lambda: workspace_status()["backend"] == "ext")
+        wait_for(lambda: state().get("available", False))
+        assert json.loads(ipc("aqueous", "status"))["session"] == session
+        if not args.force_ext:
+            wait_for(lambda: workspace_status()["backend"] == "aqueous")
+            before = [r["number"] for r in workspace_status()["rows"] if r["active"]]
+            assert ipc("aqueousTest", "workspaceReplay") == "REQUESTED"
+            time.sleep(.2)
+            assert [r["number"] for r in workspace_status()["rows"] if r["active"]] == before, "cached row survived backend generation change"
+        restarted_pids = watchers()
+        assert len(restarted_pids) == 1 and restarted_pids != watch_pids
+        watch_pids += restarted_pids
+        time.sleep(1)
+        assert state()["available"], "quiet watch became unavailable"
+        for client in clients:
+            assert command("window.close", id=client["id"])["status"] == "accepted"
+        wait_for(lambda: not [w for w in entities("window") if w["app_id"] == "aq-shell-test"])
+        # Qt exits when its Wayland connection closes; it cannot answer another IPC call after logout.
+        assert ipc("aqueousTest", "request", "aqueous.command", json.dumps(dict(action="session.exit", session=session))) == "REQUESTED"
+        assert compositor.wait(timeout=10) == 0
+        os.killpg(dms.pid, signal.SIGTERM)
+        dms.wait(timeout=10)
+        assert not any(Path("/proc", pid).exists() for pid in watch_pids), "watcher survived DMS shutdown"
+        print("PASS: daemon/UI, duplicate identities, keyboard, overview, screenshots, helper conflicts, keybinds, single watcher, workspace widget and orderly exit", flush=True)
+    finally:
+        for child in reversed(children):
+            if child.poll() is None:
+                os.killpg(child.pid, signal.SIGTERM)
+        for child in reversed(children):
+            try:
+                child.wait(timeout=8)
+            except subprocess.TimeoutExpired:
+                os.killpg(child.pid, signal.SIGKILL)
+                child.wait()
+
+
+if __name__ == "__main__":
+    main()

scripts/test-aqueous-service.py

@@ -0,0 +1,199 @@
+#!/usr/bin/env python3
+"""Exercise AqueousService's real Quickshell processes using private fake CLI fixtures."""
+
+import json
+import os
+from pathlib import Path
+import shutil
+import signal
+import subprocess
+import tempfile
+import time
+
+
+def wait_for(fn, timeout=12):
+    deadline = time.monotonic() + timeout
+    while time.monotonic() < deadline:
+        result = fn()
+        if result:
+            return result
+        time.sleep(0.05)
+    raise AssertionError("timed out waiting for QML service")
+
+
+def main():
+    root = Path(__file__).resolve().parents[1]
+    base = Path(tempfile.mkdtemp(prefix="dms-aqueous-service-"))
+    print(f"Evidence: {base}", flush=True)
+    for directory in ("Common", "Services", "bin", "runtime", "home", "config", "cache"):
+        (base / directory).mkdir(mode=0o700)
+    for name in ("AqueousService.qml", "AqueousConfigService.qml"):
+        shutil.copy(root / "quickshell/Services" / name, base / "Services" / name)
+    common = {
+        "Log": 'function scoped(name) { return {warn: function() {}}; }',
+        "I18n": 'function tr(value) { return value; }',
+        "SettingsData": 'enum Position { Top, Left, Right }',
+    }
+    services = {
+        "CompositorService": 'property bool isAqueous: false',
+        "ToastService": 'function showError(title, message) {}',
+    }
+    for directory, stubs in (("Common", common), ("Services", services)):
+        for name, body in stubs.items():
+            (base / directory / (name + ".qml")).write_text("pragma Singleton\nimport QtQuick\nQtObject { " + body + " }\n")
+    for directory in ("Common", "Services"):
+        (base / directory / "qmldir").write_text("\n".join("singleton " + path.stem + " 1.0 " + path.name for path in (base / directory).glob("*.qml")))
+    (base / "shell.qml").write_text('''import QtQuick
+import Quickshell
+import Quickshell.Io
+import qs.Services
+ShellRoot {
+    id: shell
+    property var reply: ({pending: true})
+    IpcHandler {
+        target: "test"
+        function enabled(value: bool): string { CompositorService.isAqueous = value; return "OK"; }
+        function status(): string {
+            return JSON.stringify({available: AqueousService.available, state: AqueousService.state,
+                pending: AqueousService.pendingProcesses.length, commandPending: AqueousService.commandPending,
+                retry: AqueousService.retryDelay, watching: !!AqueousService.watchProcess});
+        }
+        function command(): string {
+            shell.reply = {pending: true};
+            AqueousService.command("workspace.activate", {id: AqueousService.workspaces[0].id},
+                (ok, message) => { shell.reply = {ok: ok, message: message}; });
+            return "REQUESTED";
+        }
+        function result(): string { return JSON.stringify(shell.reply); }
+    }
+}''')
+    shutil.copy(root / "quickshell/tests/fixtures/aqueous/capabilities.json", base / "capabilities.json")
+    batch = json.loads((root / "quickshell/tests/fixtures/aqueous/watch.ndjson").read_text().splitlines()[0])
+    (base / "snapshot.json").write_text(json.dumps(batch))
+    binary = base / "bin/aqueousctl"
+    binary.write_text('''#!/usr/bin/python3
+import json, os, sys, time
+from pathlib import Path
+base = Path(os.environ["AQ_TEST_DIR"])
+args = sys.argv[1:]
+mode = (base / "mode").read_text()
+with (base / "calls").open("a") as f:
+    f.write(json.dumps({"args": args, "pid": os.getpid()}) + "\\n")
+if args[:2] == ["shell", "capabilities"]:
+    caps = json.loads((base / "capabilities.json").read_text())
+    if mode == "unsupported": caps["schema"] = 2
+    print(json.dumps(caps))
+    sys.exit(0)
+if args[:2] != ["shell", "watch"]:
+    if mode == "timeout": time.sleep(30)
+    if mode == "no-ack": sys.exit(0)
+    print(json.dumps({"ok": False, "status": "locked", "sequence": "9007199254740999"}))
+    sys.exit(1)
+batch = json.loads((base / "snapshot.json").read_text())
+batch["sequence"] = "9007199254740993"
+workspace = next(e for e in batch["upsert"] if e["kind"] == "workspace")
+workspace["name"] = "🫧 日本語"
+data = (json.dumps(batch, ensure_ascii=False) + "\\n").encode()
+split = data.index("🫧".encode()) + 1
+sys.stdout.buffer.write(data[:split]); sys.stdout.buffer.flush()
+time.sleep(.02)
+sys.stdout.buffer.write(data[split:]); sys.stdout.buffer.flush()
+delta = dict(schema=1, session=batch["session"], sequence="9007199254740999", base_sequence=batch["sequence"], type="delta", upsert=[], removed=[])
+time.sleep(.15)
+if mode == "malformed": print("{bad}", flush=True)
+elif mode == "oversize": print("x" * (4 * 1024 * 1024 + 1), flush=True)
+elif mode == "eof": sys.exit(0)
+else:
+    if mode == "mismatch": delta["base_sequence"] = "wrong"
+    # Multiple records in one write exercise coalesced reads.
+    second = dict(delta, sequence="9007199254741999", base_sequence=delta["sequence"])
+    print(json.dumps(delta) + "\\n" + json.dumps(second), flush=True)
+time.sleep(60)
+''')
+    binary.chmod(0o755)
+    env = dict(os.environ)
+    for name in ("WAYLAND_DISPLAY", "WAYLAND_SOCKET", "DISPLAY", "DMS_SOCKET"):
+        env.pop(name, None)
+    env.update(AQ_TEST_DIR=str(base), PATH=str(base / "bin") + ":/usr/bin", HOME=str(base / "home"),
+               XDG_RUNTIME_DIR=str(base / "runtime"), XDG_CONFIG_HOME=str(base / "config"), XDG_CACHE_HOME=str(base / "cache"),
+               QT_QPA_PLATFORM="offscreen", QT_QUICK_BACKEND="software")
+    (base / "mode").write_text("normal")
+    (base / "calls").touch()
+    qs = shutil.which("qs") or shutil.which("quickshell")
+
+    def ipc(*args):
+        result = subprocess.run([qs, "ipc", "-p", str(base / "shell.qml"), "call", "test", *args], env=env, capture_output=True, text=True, timeout=5)
+        assert result.returncode == 0, (result.stdout, result.stderr)
+        return result.stdout.strip()
+
+    def state():
+        return json.loads(ipc("status"))
+
+    def calls():
+        return [json.loads(line) for line in (base / "calls").read_text().splitlines()]
+
+    def reply():
+        response = json.loads(ipc("result"))
+        return response if not response.get("pending") else None
+
+    def stop():
+        ipc("enabled", "false")
+        wait_for(lambda: not state()["watching"] and not state()["pending"])
+        time.sleep(.1)
+        assert not any(Path("/proc", str(call["pid"])).exists() for call in calls()), "child process survived backend change"
+
+    with (base / "quickshell.log").open("w") as log:
+        process = subprocess.Popen([qs, "-p", str(base / "shell.qml")], env=env, stdout=log, stderr=log, start_new_session=True)
+        try:
+            wait_for(lambda: subprocess.run([qs, "ipc", "-p", str(base / "shell.qml"), "call", "test", "status"], env=env, capture_output=True).returncode == 0)
+            assert not calls(), "Aqueous helper initialized in unrelated backend"
+            ipc("enabled", "true")
+            current = wait_for(lambda: (s if (s := state())["available"] and s["state"]["batch"]["sequence"] == "9007199254741999" else None))
+            assert any(e.get("name") == "🫧 日本語" for e in current["state"]["batch"]["upsert"])
+            count = len(calls())
+            time.sleep(1)
+            assert state()["available"] and len(calls()) == count, "quiet stream spawned subprocesses"
+            ipc("command")
+            assert "locked" in wait_for(reply)["message"]
+            for mode in ("malformed", "mismatch", "oversize", "eof", "unsupported"):
+                stop()
+                (base / "mode").write_text(mode)
+                ipc("enabled", "true")
+                wait_for(lambda: state()["state"].get("error"))
+                assert not state()["available"], mode
+                assert state()["retry"] >= 1000
+            stop()
+            binary.rename(base / "bin/aqueousctl-disabled")
+            ipc("enabled", "true")
+            wait_for(lambda: state()["state"].get("error"))
+            assert not state()["pending"]
+            time.sleep(.7)
+            assert state()["retry"] <= 4000, "missing CLI caused a tight retry loop"
+            stop()
+            (base / "bin/aqueousctl-disabled").rename(binary)
+            (base / "mode").write_text("no-ack")
+            ipc("enabled", "true")
+            wait_for(lambda: state()["available"])
+            ipc("command")
+            assert not wait_for(reply)["ok"], "EOF without acknowledgement accepted"
+            stop()
+            (base / "mode").write_text("timeout")
+            ipc("enabled", "true")
+            wait_for(lambda: state()["available"])
+            count = len([c for c in calls() if c["args"][0] == "workspace"])
+            ipc("command")
+            assert "uncertain" in wait_for(reply)["message"]
+            assert len([c for c in calls() if c["args"][0] == "workspace"]) == count + 1, "timed-out mutation retried"
+            stop()
+            (base / "mode").write_text("normal")
+            ipc("enabled", "true")
+            wait_for(lambda: state()["available"])
+        finally:
+            os.killpg(process.pid, signal.SIGTERM)
+            process.wait(timeout=5)
+    wait_for(lambda: not any(Path("/proc", str(call["pid"])).exists() for call in calls()))
+    print("PASS: QML process framing, Unicode, continuity, errors, retry, timeout, idle and cleanup", flush=True)
+
+
+if __name__ == "__main__":
+    main()