This repository has no description
7.2 kB
211 lines
1#!/usr/bin/env python3
2"""Place new sway windows on the workspace of the process that spawned them.
3
4Sway maps a new view onto the *currently focused* workspace. That is wrong
5whenever the spawning process lives elsewhere: a build finishing in a terminal
6on workspace "strom" pops its window onto whatever workspace happens to be
7focused at that moment.
8
9This daemon subscribes to sway's window::new events and resolves the workspace
10of the spawning process by walking /proc PPid links until it hits a pid that
11already owns a window. Terminals here exec tmux, which breaks the ppid chain at
12the (session-global) tmux server, so the walk also hops tmux pane -> session ->
13attached client pid and continues from there.
14
15Windows whose ancestry cannot be resolved (systemd user services, dbus
16activation, single-process multi-window apps like Firefox) keep sway's default
17behaviour.
18"""
19
20from __future__ import annotations
21
22import json
23import subprocess
24import sys
25from collections.abc import Iterator
26from typing import Any
27
28# Sway's hidden scratchpad workspace; views parked there must never be moved,
29# and a view living there is not a placement hint for anything else.
30SCRATCHPAD = "__i3_scratch"
31
32
33def run(args: list[str]) -> str | None:
34 """Run a command, returning its stdout, or None when it failed."""
35 try:
36 proc = subprocess.run(args, capture_output=True, text=True, check=False)
37 except OSError as e:
38 print(f"failed to run {args[0]}: {e}", file=sys.stderr)
39 return None
40 if proc.returncode != 0:
41 return None
42 return proc.stdout
43
44
45def swaymsg_json(*args: str) -> Any:
46 """Query sway IPC, returning the parsed reply or None."""
47 out = run(["swaymsg", "-r", "-t", *args])
48 if out is None:
49 return None
50 try:
51 return json.loads(out)
52 except json.JSONDecodeError as e:
53 print(f"failed to parse sway reply: {e}", file=sys.stderr)
54 return None
55
56
57def iter_views(
58 node: dict[str, Any], workspace: str | None
59) -> Iterator[tuple[str, int, int]]:
60 """Yield (workspace name, container id, pid) for every view in the tree."""
61 if node.get("type") == "workspace":
62 workspace = node.get("name")
63 pid = node.get("pid")
64 if pid is not None and workspace is not None:
65 yield workspace, int(node["id"]), int(pid)
66 for key in ("nodes", "floating_nodes"):
67 for child in node.get(key, []):
68 yield from iter_views(child, workspace)
69
70
71def parent_pid(pid: int) -> int | None:
72 """Read PPid from /proc, or None if the process is gone."""
73 try:
74 with open(f"/proc/{pid}/status", encoding="utf-8") as f:
75 for line in f:
76 if line.startswith("PPid:"):
77 return int(line.split()[1])
78 except (OSError, ValueError):
79 return None
80 return None
81
82
83def tmux_hops() -> dict[int, list[int]]:
84 """Map each tmux pane's root pid to the pids of its session's clients.
85
86 A process started inside a pane has that pane's root pid as an ancestor,
87 but its ppid chain leads to the tmux server, not to the terminal window.
88 The attached client processes are children of that terminal, so they are
89 where the walk has to continue.
90 """
91 panes = run(["tmux", "list-panes", "-a", "-F", "#{pane_pid} #{session_id}"])
92 clients = run(["tmux", "list-clients", "-F", "#{client_pid} #{session_id}"])
93 if not panes or not clients:
94 return {}
95 session_clients: dict[str, list[int]] = {}
96 for line in clients.splitlines():
97 pid, _, session = line.partition(" ")
98 try:
99 session_clients.setdefault(session, []).append(int(pid))
100 except ValueError:
101 continue
102 hops: dict[int, list[int]] = {}
103 for line in panes.splitlines():
104 pid, _, session = line.partition(" ")
105 try:
106 hops[int(pid)] = session_clients.get(session, [])
107 except ValueError:
108 continue
109 return hops
110
111
112def resolve_workspace(
113 pid: int,
114 pid_workspaces: dict[int, str],
115 hops: dict[int, list[int]],
116 visited: set[int],
117) -> str | None:
118 """Find the workspace of the nearest ancestor process that owns a window."""
119 current: int | None = pid
120 while current is not None and current > 1:
121 if current in visited:
122 return None
123 visited.add(current)
124 workspace = pid_workspaces.get(current)
125 if workspace is not None:
126 return workspace
127 for client in hops.get(current, []):
128 workspace = resolve_workspace(client, pid_workspaces, hops, visited)
129 if workspace is not None:
130 return workspace
131 current = parent_pid(current)
132 return None
133
134
135def quote(name: str) -> str:
136 """Quote a workspace name for a sway command."""
137 escaped = name.replace("\\", "\\\\").replace('"', '\\"')
138 return f'"{escaped}"'
139
140
141def handle_new_window(container_id: int, pid: int) -> None:
142 """Move a freshly mapped view to the workspace of its spawning process."""
143 tree = swaymsg_json("get_tree")
144 if tree is None:
145 return
146
147 origin: str | None = None
148 pid_workspaces: dict[int, str] = {}
149 for workspace, view_id, view_pid in iter_views(tree, None):
150 if view_id == container_id:
151 origin = workspace
152 continue
153 # Views of the same process are no hint: a single-process multi-window
154 # app (Firefox) would drag every new window to an unrelated workspace.
155 if view_pid == pid or workspace.startswith(SCRATCHPAD):
156 continue
157 pid_workspaces.setdefault(view_pid, workspace)
158
159 if origin is None or origin.startswith(SCRATCHPAD):
160 # Window already closed again, or parked in the scratchpad by a
161 # for_window rule (the workspace-manager Firefox anchor).
162 return
163
164 target = resolve_workspace(pid, pid_workspaces, tmux_hops(), set())
165 if target is None or target == origin:
166 return
167
168 print(
169 f"moving container {container_id} (pid {pid}): {origin} -> {target}",
170 file=sys.stderr,
171 )
172 run(
173 [
174 "swaymsg",
175 "--",
176 f"[con_id={container_id}] move container to workspace {quote(target)}",
177 ]
178 )
179
180
181def main() -> int:
182 """Subscribe to sway window events and place new windows."""
183 proc = subprocess.Popen(
184 ["swaymsg", "-t", "subscribe", "-m", '["window"]'],
185 stdout=subprocess.PIPE,
186 text=True,
187 )
188 if proc.stdout is None:
189 print("no stdout from swaymsg subscribe", file=sys.stderr)
190 return 1
191
192 for line in proc.stdout:
193 try:
194 event: dict[str, Any] = json.loads(line)
195 except json.JSONDecodeError as e:
196 print(f"failed to parse sway event: {e}", file=sys.stderr)
197 continue
198 if event.get("change") != "new":
199 continue
200 container: dict[str, Any] = event.get("container") or {}
201 pid = container.get("pid")
202 container_id = container.get("id")
203 if pid is None or container_id is None:
204 continue
205 handle_new_window(int(container_id), int(pid))
206
207 return proc.wait()
208
209
210if __name__ == "__main__":
211 sys.exit(main())