feat: fix super+a shortcut

modules/capabilities/niri/_shortcuts.nix

@@ -1,37 +1,88 @@
 { pkgs }:
-pkgs.writeShellApplication {
+pkgs.writeScriptBin "niri-shortcut-swap-monitor-workspaces" ''
-  name = "niri-shortcut-swap-monitor-workspaces";
+  #!${pkgs.python3}/bin/python3
-  runtimeInputs = [
+  import fcntl
-    pkgs.jq
+  import json
-    pkgs.niri
+  import os
-  ];
+  import socket
-  text = ''
+  import sys
-    focused_output="$(niri msg --json focused-output | jq -r '.name // empty')"
-    if [ -z "$focused_output" ]; then
-      exit 0
-    fi
 
-    niri msg action focus-monitor-next
 
-    target_output="$(niri msg --json focused-output | jq -r '.name // empty')"
+  def disposable(workspace):
-    if [ -z "$target_output" ] || [ "$target_output" = "$focused_output" ]; then
+      return workspace["active_window_id"] is None and workspace["name"] is None
-      exit 0
-    fi
 
-    target_workspace_idx="$(
-      niri msg --json workspaces \
-        | jq -r --arg output "$target_output" '
-            first(.[] | select(.output == $output and .is_active) | .idx) // empty
-          '
-    )"
-    if [ -z "$target_workspace_idx" ]; then
-      niri msg action focus-monitor "$focused_output"
-      exit 0
-    fi
 
-    niri msg action focus-monitor "$focused_output"
+  def swap(connection):
-    niri msg action move-workspace-to-monitor "$target_output"
+      with connection.makefile("rw", encoding="utf-8") as stream:
-    niri msg action focus-workspace "$target_workspace_idx"
+          def request(message):
-    niri msg action move-workspace-to-monitor "$focused_output"
+              stream.write(json.dumps(message) + "\n")
-  '';
+              stream.flush()
-}
+              reply = json.loads(stream.readline())
+              if "Err" in reply:
+                  raise RuntimeError(reply["Err"])
+              return reply["Ok"]
+
+          def action(name, **arguments):
+              request({"Action": {name: arguments}})
+
+          workspaces = request("Workspaces")["Workspaces"]
+          source = next((ws for ws in workspaces if ws["is_focused"]), None)
+          active = {
+              ws["output"]: ws for ws in workspaces
+              if ws["output"] is not None and ws["is_active"]
+          }
+          if source is None or source["output"] not in active or len(active) < 2:
+              return
+          outputs = sorted(active)
+          target = active[outputs[(outputs.index(source["output"]) + 1) % len(outputs)]]
+          if source["active_window_id"] is None and target["active_window_id"] is None:
+              return
+
+          # IDs survive moves; numeric workspace positions do not.
+          moves = ((source, target), (target, source))
+          for workspace, destination in moves:
+              action(
+                  "MoveWorkspaceToMonitor",
+                  reference={"Id": workspace["id"]},
+                  output=destination["output"],
+              )
+
+          # Activate the original monitor last to keep keyboard focus there.
+          for workspace, destination in moves:
+              current = request("Workspaces")["Workspaces"]
+              arriving = next((
+                  ws for ws in current
+                  if ws["id"] == workspace["id"] and ws["output"] == destination["output"]
+              ), None)
+              if arriving is None and disposable(workspace):
+                  # Niri may discard an unnamed empty workspace during the swap.
+                  arriving = max((
+                      ws for ws in current
+                      if ws["output"] == destination["output"] and disposable(ws)
+                  ), key=lambda ws: ws["idx"], default=None)
+              if arriving is None:
+                  raise RuntimeError("workspace or monitor disappeared during swap")
+              reference = {"Id": arriving["id"]}
+              # Focus empty workspaces before reordering so Niri retains them.
+              action("FocusWorkspace", reference=reference)
+              action("MoveWorkspaceToIndex", reference=reference, index=destination["idx"])
+
+
+  def main():
+      lock_path = os.path.join(os.environ["XDG_RUNTIME_DIR"], "niri-swap-workspaces.lock")
+      with open(lock_path, "a") as lock:
+          try:
+              fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
+          except BlockingIOError:
+              return
+          with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as connection:
+              connection.settimeout(5)
+              connection.connect(os.environ["NIRI_SOCKET"])
+              swap(connection)
+
+
+  try:
+      main()
+  except (OSError, ValueError, KeyError, RuntimeError) as error:
+      sys.exit(f"niri workspace swap: {error}")
+''