This repository has no description
12 kB
349 lines
1#!/usr/bin/env python3
2"""
3Workspace Manager Daemon
4
5A compositor-agnostic workspace manager that:
6- Subscribes to sway IPC for workspace change events
7- Runs user-defined hooks on workspace enter/leave
8- Provides current workspace and directory info via Unix socket
9
10Config merging: user config is loaded first, then system config (declarative) overrides per-field
11"""
12
13from __future__ import annotations
14
15import asyncio
16import json
17import os
18import signal
19import subprocess
20import sys
21from pathlib import Path
22from typing import Any
23
24# XDG runtime directory for socket and state
25RUNTIME_DIR = Path(os.environ.get("XDG_RUNTIME_DIR", f"/run/user/{os.getuid()}"))
26SOCKET_DIR = RUNTIME_DIR / "workspace-manager"
27SOCKET_PATH = SOCKET_DIR / "socket"
28STATE_PATH = SOCKET_DIR / "state.json"
29
30# System config directory (declarative, takes precedence)
31_system_config_dir_env = os.environ.get("WORKSPACE_MANAGER_SYSTEM_CONFIG_DIR")
32SYSTEM_CONFIG_DIR: Path | None = (
33 Path(_system_config_dir_env).expanduser() if _system_config_dir_env else None
34)
35
36# User config directory (fallback)
37_config_dir_env = os.environ.get("WORKSPACE_MANAGER_CONFIG_DIR")
38CONFIG_DIR = (
39 Path(_config_dir_env).expanduser()
40 if _config_dir_env
41 else Path.home() / "workspaces"
42)
43
44_default_dir_env = os.environ.get("WORKSPACE_MANAGER_DEFAULT_DIR")
45DEFAULT_DIRECTORY = (
46 Path(_default_dir_env).expanduser() if _default_dir_env else Path.home()
47)
48
49
50def load_config_file(config_file: Path) -> dict[str, Any] | None:
51 """Load a workspace config file, returning None if it doesn't exist or fails."""
52 if not config_file.exists():
53 return None
54 try:
55 with open(config_file) as f:
56 return json.load(f) # type: ignore[no-any-return]
57 except (json.JSONDecodeError, OSError) as e:
58 print(f"Warning: Failed to load config {config_file}: {e}", file=sys.stderr)
59 return None
60
61
62class WorkspaceConfig:
63 """Configuration for a workspace.
64
65 Merges user config with system config (declarative).
66 System config fields take precedence over user config fields.
67 """
68
69 def __init__(self, name: str) -> None:
70 self.name = name
71 self.directory: Path = DEFAULT_DIRECTORY
72 self.on_enter: list[str] = []
73 self.on_leave: list[str] = []
74 self._load()
75
76 def _load(self) -> None:
77 # Load user config first (base)
78 user_config_file = CONFIG_DIR / f"{self.name}.json"
79 user_data = load_config_file(user_config_file)
80
81 # Load system config (declarative, takes precedence)
82 system_data: dict[str, Any] | None = None
83 if SYSTEM_CONFIG_DIR is not None:
84 system_config_file = SYSTEM_CONFIG_DIR / f"{self.name}.json"
85 system_data = load_config_file(system_config_file)
86
87 # Merge: start with user config, override with system config
88 merged: dict[str, Any] = {}
89 if user_data is not None:
90 merged.update(user_data)
91 if system_data is not None:
92 merged.update(system_data)
93
94 # Apply merged config
95 if "directory" in merged:
96 self.directory = Path(str(merged["directory"])).expanduser()
97 if "on_enter" in merged:
98 self.on_enter = list(merged["on_enter"])
99 if "on_leave" in merged:
100 self.on_leave = list(merged["on_leave"])
101
102
103class WorkspaceManager:
104 """Manages workspace state and configuration"""
105
106 def __init__(self) -> None:
107 self.current_workspace: str | None = None
108 self.configs: dict[str, WorkspaceConfig] = {}
109
110 def get_config(self, workspace: str) -> WorkspaceConfig:
111 """Get or create config for a workspace (configs are loaded fresh each time)"""
112 # Always reload config to pick up changes
113 self.configs[workspace] = WorkspaceConfig(workspace)
114 return self.configs[workspace]
115
116 def get_current_directory(self) -> Path:
117 """Get the directory for the current workspace"""
118 if self.current_workspace:
119 config = self.get_config(self.current_workspace)
120 return config.directory
121 return DEFAULT_DIRECTORY
122
123 async def handle_workspace_change(
124 self, old_workspace: str | None, new_workspace: str
125 ) -> None:
126 """Handle a workspace focus change event"""
127 # Run on_leave hooks for old workspace
128 if old_workspace:
129 old_config = self.get_config(old_workspace)
130 for hook in old_config.on_leave:
131 await self._run_hook(hook, "on_leave", old_workspace)
132
133 # Update current workspace
134 self.current_workspace = new_workspace
135
136 # Run on_enter hooks for new workspace
137 new_config = self.get_config(new_workspace)
138 for hook in new_config.on_enter:
139 await self._run_hook(hook, "on_enter", new_workspace)
140
141 # Write state file
142 self._write_state()
143
144 async def _run_hook(self, command: str, hook_type: str, workspace: str) -> None:
145 """Run a hook command in the background"""
146 try:
147 # Run in background, don't wait
148 subprocess.Popen(
149 command,
150 shell=True,
151 stdout=subprocess.DEVNULL,
152 stderr=subprocess.DEVNULL,
153 start_new_session=True,
154 )
155 print(f"Ran {hook_type} hook for {workspace}: {command}", file=sys.stderr)
156 except OSError as e:
157 print(
158 f"Failed to run {hook_type} hook for {workspace}: {e}",
159 file=sys.stderr,
160 )
161
162 def _write_state(self) -> None:
163 """Write current state to state file"""
164 state: dict[str, Any] = {
165 "workspace": self.current_workspace,
166 "directory": str(self.get_current_directory()),
167 }
168 try:
169 STATE_PATH.parent.mkdir(parents=True, exist_ok=True)
170 with open(STATE_PATH, "w") as f:
171 json.dump(state, f)
172 except OSError as e:
173 print(f"Failed to write state file: {e}", file=sys.stderr)
174
175
176manager = WorkspaceManager()
177
178
179async def subscribe_sway() -> None:
180 """Subscribe to sway IPC workspace events"""
181 print("Starting sway workspace subscription...", file=sys.stderr)
182
183 while True:
184 try:
185 proc = await asyncio.create_subprocess_exec(
186 "swaymsg",
187 "-t",
188 "subscribe",
189 "-m",
190 '["workspace"]',
191 stdout=asyncio.subprocess.PIPE,
192 stderr=asyncio.subprocess.PIPE,
193 )
194
195 if proc.stdout is None:
196 print("Failed to get stdout from swaymsg", file=sys.stderr)
197 await asyncio.sleep(5)
198 continue
199
200 async for line in proc.stdout:
201 try:
202 event: dict[str, Any] = json.loads(line.decode().strip())
203 if event.get("change") == "focus":
204 current: dict[str, Any] = event.get("current", {})
205 old: dict[str, Any] | None = event.get("old")
206 new_name: str | None = current.get("name")
207 old_name: str | None = old.get("name") if old else None
208
209 if new_name:
210 print(
211 f"Workspace change: {old_name} -> {new_name}",
212 file=sys.stderr,
213 )
214 await manager.handle_workspace_change(old_name, new_name)
215 except json.JSONDecodeError as e:
216 print(f"Failed to parse sway event: {e}", file=sys.stderr)
217
218 # If swaymsg exits, wait a bit and retry
219 await proc.wait()
220 print("swaymsg exited, retrying in 5 seconds...", file=sys.stderr)
221 await asyncio.sleep(5)
222
223 except FileNotFoundError:
224 print("swaymsg not found, retrying in 10 seconds...", file=sys.stderr)
225 await asyncio.sleep(10)
226 except Exception as e:
227 print(
228 f"Error in sway subscription: {e}, retrying in 5 seconds...",
229 file=sys.stderr,
230 )
231 await asyncio.sleep(5)
232
233
234async def handle_client(
235 reader: asyncio.StreamReader, writer: asyncio.StreamWriter
236) -> None:
237 """Handle a client connection to the Unix socket"""
238 try:
239 data = await reader.read(1024)
240 command = data.decode().strip()
241
242 response: dict[str, Any]
243 if command == "dir":
244 response = {"directory": str(manager.get_current_directory())}
245 elif command == "workspace":
246 response = {"workspace": manager.current_workspace}
247 elif command == "status":
248 response = {
249 "workspace": manager.current_workspace,
250 "directory": str(manager.get_current_directory()),
251 "running": True,
252 }
253 else:
254 response = {"error": f"Unknown command: {command}"}
255
256 writer.write(json.dumps(response).encode() + b"\n")
257 await writer.drain()
258 except Exception as e:
259 print(f"Error handling client: {e}", file=sys.stderr)
260 finally:
261 writer.close()
262 await writer.wait_closed()
263
264
265async def socket_server() -> None:
266 """Run the Unix socket server for CLI queries"""
267 # Ensure socket directory exists
268 SOCKET_DIR.mkdir(parents=True, exist_ok=True)
269
270 # Remove stale socket if exists
271 if SOCKET_PATH.exists():
272 SOCKET_PATH.unlink()
273
274 server = await asyncio.start_unix_server(handle_client, path=str(SOCKET_PATH))
275
276 # Make socket accessible
277 os.chmod(SOCKET_PATH, 0o600)
278
279 print(f"Socket server listening on {SOCKET_PATH}", file=sys.stderr)
280
281 async with server:
282 await server.serve_forever()
283
284
285async def get_current_workspace() -> str | None:
286 """Get the current workspace from sway"""
287 try:
288 proc = await asyncio.create_subprocess_exec(
289 "swaymsg",
290 "-t",
291 "get_workspaces",
292 stdout=asyncio.subprocess.PIPE,
293 stderr=asyncio.subprocess.PIPE,
294 )
295 stdout, _ = await proc.communicate()
296 workspaces: list[dict[str, Any]] = json.loads(stdout.decode())
297 for ws in workspaces:
298 if ws.get("focused"):
299 name = ws.get("name")
300 return str(name) if name is not None else None
301 except Exception as e:
302 print(f"Failed to get current workspace: {e}", file=sys.stderr)
303 return None
304
305
306async def main() -> None:
307 """Main entry point"""
308 # Ensure user config directory exists (system config is read-only from nix store)
309 CONFIG_DIR.mkdir(parents=True, exist_ok=True)
310
311 # Log config directories
312 if SYSTEM_CONFIG_DIR:
313 print(f"System config dir: {SYSTEM_CONFIG_DIR}", file=sys.stderr)
314 print(f"User config dir: {CONFIG_DIR}", file=sys.stderr)
315
316 # Get initial workspace
317 initial_workspace = await get_current_workspace()
318 if initial_workspace:
319 manager.current_workspace = initial_workspace
320 manager._write_state()
321 print(f"Initial workspace: {initial_workspace}", file=sys.stderr)
322
323 # Handle shutdown gracefully
324 loop = asyncio.get_event_loop()
325
326 def shutdown(sig: signal.Signals) -> None:
327 print(f"Received {sig.name}, shutting down...", file=sys.stderr)
328 for task in asyncio.all_tasks(loop):
329 task.cancel()
330
331 for sig in (signal.SIGTERM, signal.SIGINT):
332 loop.add_signal_handler(sig, shutdown, sig)
333
334 # Run both tasks concurrently
335 try:
336 await asyncio.gather(
337 subscribe_sway(),
338 socket_server(),
339 )
340 except asyncio.CancelledError:
341 print("Shutdown complete", file=sys.stderr)
342 finally:
343 # Cleanup socket
344 if SOCKET_PATH.exists():
345 SOCKET_PATH.unlink()
346
347
348if __name__ == "__main__":
349 asyncio.run(main())