This repository has no description
16 kB
445 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
49TERMINAL_COMMAND = os.environ.get("WORKSPACE_MANAGER_TERMINAL", "kitty")
50
51
52def load_config_file(config_file: Path) -> dict[str, Any] | None:
53 """Load a workspace config file, returning None if it doesn't exist or fails."""
54 if not config_file.exists():
55 return None
56 try:
57 with open(config_file) as f:
58 return json.load(f) # type: ignore[no-any-return]
59 except (json.JSONDecodeError, OSError) as e:
60 print(f"Warning: Failed to load config {config_file}: {e}", file=sys.stderr)
61 return None
62
63
64class WorkspaceConfig:
65 """Configuration for a workspace.
66
67 Merges user config with system config (declarative).
68 System config fields take precedence over user config fields.
69 """
70
71 def __init__(self, name: str) -> None:
72 self.name = name
73 self.directory: Path = DEFAULT_DIRECTORY
74 self.on_enter: list[str] = []
75 self.on_leave: list[str] = []
76 self.on_create: list[str] = []
77 self._load()
78
79 def _load(self) -> None:
80 # Load user config first (base)
81 user_config_file = CONFIG_DIR / f"{self.name}.json"
82 user_data = load_config_file(user_config_file)
83
84 # Load system config (declarative, takes precedence)
85 system_data: dict[str, Any] | None = None
86 if SYSTEM_CONFIG_DIR is not None:
87 system_config_file = SYSTEM_CONFIG_DIR / f"{self.name}.json"
88 system_data = load_config_file(system_config_file)
89
90 # Merge: start with user config, override with system config
91 merged: dict[str, Any] = {}
92 if user_data is not None:
93 merged.update(user_data)
94 if system_data is not None:
95 merged.update(system_data)
96
97 # Apply merged config
98 if "directory" in merged:
99 self.directory = Path(str(merged["directory"])).expanduser()
100 if "on_enter" in merged:
101 self.on_enter = list(merged["on_enter"])
102 if "on_leave" in merged:
103 self.on_leave = list(merged["on_leave"])
104 if "on_create" in merged:
105 self.on_create = list(merged["on_create"])
106
107
108class WorkspaceManager:
109 """Manages workspace state and configuration"""
110
111 def __init__(self) -> None:
112 self.current_workspace: str | None = None
113 self.configs: dict[str, WorkspaceConfig] = {}
114 self.created_workspaces: set[str] = set()
115
116 def get_config(self, workspace: str) -> WorkspaceConfig:
117 """Get or create config for a workspace (configs are loaded fresh each time)"""
118 # Always reload config to pick up changes
119 self.configs[workspace] = WorkspaceConfig(workspace)
120 return self.configs[workspace]
121
122 def get_current_directory(self) -> Path:
123 """Get the directory for the current workspace"""
124 if self.current_workspace:
125 config = self.get_config(self.current_workspace)
126 return config.directory
127 return DEFAULT_DIRECTORY
128
129 def _tmux_session_exists(self, session_name: str) -> bool:
130 """Check if a tmux session already exists"""
131 try:
132 result = subprocess.run(
133 ["tmux", "has-session", "-t", session_name],
134 capture_output=True,
135 )
136 return result.returncode == 0
137 except FileNotFoundError:
138 print("Warning: tmux not found", file=sys.stderr)
139 return False
140
141 async def _run_on_create(self, workspace: str, config: WorkspaceConfig) -> None:
142 """Start on_create commands each in their own tmux session + terminal.
143
144 For each command, creates a tmux session named <workspace>-start<N>
145 running the command, then opens a terminal window attached to it.
146 If the terminal is closed, reattach with: tmux attach -t <workspace>-start<N>
147 """
148 if not config.on_create:
149 return
150
151 work_dir = str(config.directory)
152
153 for i, cmd in enumerate(config.on_create, start=1):
154 session_name = f"{workspace}-start{i}"
155
156 # Skip if tmux session already exists (e.g. from a previous run)
157 if self._tmux_session_exists(session_name):
158 print(
159 f"Tmux session '{session_name}' already exists, skipping",
160 file=sys.stderr,
161 )
162 continue
163
164 # Create a detached tmux session running the command
165 try:
166 subprocess.run(
167 [
168 "tmux",
169 "new-session",
170 "-d",
171 "-s",
172 session_name,
173 "-c",
174 work_dir,
175 cmd,
176 ],
177 check=True,
178 )
179 print(
180 f"Created tmux session '{session_name}': {cmd}",
181 file=sys.stderr,
182 )
183 except (OSError, subprocess.CalledProcessError) as e:
184 print(
185 f"Failed to create tmux session '{session_name}': {e}",
186 file=sys.stderr,
187 )
188 continue
189
190 # Open a terminal window attached to the tmux session
191 try:
192 subprocess.Popen(
193 [
194 TERMINAL_COMMAND,
195 "-T",
196 session_name,
197 "tmux",
198 "attach",
199 "-t",
200 session_name,
201 ],
202 start_new_session=True,
203 )
204 print(
205 f"Opened terminal for tmux session '{session_name}'",
206 file=sys.stderr,
207 )
208 except OSError as e:
209 print(
210 f"Failed to open terminal for '{session_name}': {e}",
211 file=sys.stderr,
212 )
213
214 async def handle_workspace_change(
215 self, old_workspace: str | None, new_workspace: str
216 ) -> None:
217 """Handle a workspace focus change event"""
218 # Run on_leave hooks for old workspace
219 if old_workspace:
220 old_config = self.get_config(old_workspace)
221 for hook in old_config.on_leave:
222 await self._run_hook(hook, "on_leave", old_workspace)
223
224 # Update current workspace
225 self.current_workspace = new_workspace
226
227 # Run on_create hooks if this is the first time entering the workspace
228 new_config = self.get_config(new_workspace)
229 if new_workspace not in self.created_workspaces:
230 self.created_workspaces.add(new_workspace)
231 await self._run_on_create(new_workspace, new_config)
232
233 # Run on_enter hooks for new workspace
234 for hook in new_config.on_enter:
235 await self._run_hook(hook, "on_enter", new_workspace)
236
237 # Write state file
238 self._write_state()
239
240 async def _run_hook(self, command: str, hook_type: str, workspace: str) -> None:
241 """Run a hook command in the background"""
242 try:
243 # Run in background, don't wait
244 subprocess.Popen(
245 command,
246 shell=True,
247 stdout=subprocess.DEVNULL,
248 stderr=subprocess.DEVNULL,
249 start_new_session=True,
250 )
251 print(f"Ran {hook_type} hook for {workspace}: {command}", file=sys.stderr)
252 except OSError as e:
253 print(
254 f"Failed to run {hook_type} hook for {workspace}: {e}",
255 file=sys.stderr,
256 )
257
258 def _write_state(self) -> None:
259 """Write current state to state file"""
260 state: dict[str, Any] = {
261 "workspace": self.current_workspace,
262 "directory": str(self.get_current_directory()),
263 }
264 try:
265 STATE_PATH.parent.mkdir(parents=True, exist_ok=True)
266 with open(STATE_PATH, "w") as f:
267 json.dump(state, f)
268 except OSError as e:
269 print(f"Failed to write state file: {e}", file=sys.stderr)
270
271
272manager = WorkspaceManager()
273
274
275async def subscribe_sway() -> None:
276 """Subscribe to sway IPC workspace events"""
277 print("Starting sway workspace subscription...", file=sys.stderr)
278
279 while True:
280 try:
281 proc = await asyncio.create_subprocess_exec(
282 "swaymsg",
283 "-t",
284 "subscribe",
285 "-m",
286 '["workspace"]',
287 stdout=asyncio.subprocess.PIPE,
288 stderr=asyncio.subprocess.PIPE,
289 )
290
291 if proc.stdout is None:
292 print("Failed to get stdout from swaymsg", file=sys.stderr)
293 await asyncio.sleep(5)
294 continue
295
296 async for line in proc.stdout:
297 try:
298 event: dict[str, Any] = json.loads(line.decode().strip())
299 if event.get("change") == "focus":
300 current: dict[str, Any] = event.get("current", {})
301 old: dict[str, Any] | None = event.get("old")
302 new_name: str | None = current.get("name")
303 old_name: str | None = old.get("name") if old else None
304
305 if new_name:
306 print(
307 f"Workspace change: {old_name} -> {new_name}",
308 file=sys.stderr,
309 )
310 await manager.handle_workspace_change(old_name, new_name)
311 except json.JSONDecodeError as e:
312 print(f"Failed to parse sway event: {e}", file=sys.stderr)
313
314 # If swaymsg exits, wait a bit and retry
315 await proc.wait()
316 print("swaymsg exited, retrying in 5 seconds...", file=sys.stderr)
317 await asyncio.sleep(5)
318
319 except FileNotFoundError:
320 print("swaymsg not found, retrying in 10 seconds...", file=sys.stderr)
321 await asyncio.sleep(10)
322 except Exception as e:
323 print(
324 f"Error in sway subscription: {e}, retrying in 5 seconds...",
325 file=sys.stderr,
326 )
327 await asyncio.sleep(5)
328
329
330async def handle_client(
331 reader: asyncio.StreamReader, writer: asyncio.StreamWriter
332) -> None:
333 """Handle a client connection to the Unix socket"""
334 try:
335 data = await reader.read(1024)
336 command = data.decode().strip()
337
338 response: dict[str, Any]
339 if command == "dir":
340 response = {"directory": str(manager.get_current_directory())}
341 elif command == "workspace":
342 response = {"workspace": manager.current_workspace}
343 elif command == "status":
344 response = {
345 "workspace": manager.current_workspace,
346 "directory": str(manager.get_current_directory()),
347 "running": True,
348 }
349 else:
350 response = {"error": f"Unknown command: {command}"}
351
352 writer.write(json.dumps(response).encode() + b"\n")
353 await writer.drain()
354 except Exception as e:
355 print(f"Error handling client: {e}", file=sys.stderr)
356 finally:
357 writer.close()
358 await writer.wait_closed()
359
360
361async def socket_server() -> None:
362 """Run the Unix socket server for CLI queries"""
363 # Ensure socket directory exists
364 SOCKET_DIR.mkdir(parents=True, exist_ok=True)
365
366 # Remove stale socket if exists
367 if SOCKET_PATH.exists():
368 SOCKET_PATH.unlink()
369
370 server = await asyncio.start_unix_server(handle_client, path=str(SOCKET_PATH))
371
372 # Make socket accessible
373 os.chmod(SOCKET_PATH, 0o600)
374
375 print(f"Socket server listening on {SOCKET_PATH}", file=sys.stderr)
376
377 async with server:
378 await server.serve_forever()
379
380
381async def get_current_workspace() -> str | None:
382 """Get the current workspace from sway"""
383 try:
384 proc = await asyncio.create_subprocess_exec(
385 "swaymsg",
386 "-t",
387 "get_workspaces",
388 stdout=asyncio.subprocess.PIPE,
389 stderr=asyncio.subprocess.PIPE,
390 )
391 stdout, _ = await proc.communicate()
392 workspaces: list[dict[str, Any]] = json.loads(stdout.decode())
393 for ws in workspaces:
394 if ws.get("focused"):
395 name = ws.get("name")
396 return str(name) if name is not None else None
397 except Exception as e:
398 print(f"Failed to get current workspace: {e}", file=sys.stderr)
399 return None
400
401
402async def main() -> None:
403 """Main entry point"""
404 # Ensure user config directory exists (system config is read-only from nix store)
405 CONFIG_DIR.mkdir(parents=True, exist_ok=True)
406
407 # Log config directories
408 if SYSTEM_CONFIG_DIR:
409 print(f"System config dir: {SYSTEM_CONFIG_DIR}", file=sys.stderr)
410 print(f"User config dir: {CONFIG_DIR}", file=sys.stderr)
411
412 # Get initial workspace
413 initial_workspace = await get_current_workspace()
414 if initial_workspace:
415 manager.current_workspace = initial_workspace
416 manager._write_state()
417 print(f"Initial workspace: {initial_workspace}", file=sys.stderr)
418
419 # Handle shutdown gracefully
420 loop = asyncio.get_event_loop()
421
422 def shutdown(sig: signal.Signals) -> None:
423 print(f"Received {sig.name}, shutting down...", file=sys.stderr)
424 for task in asyncio.all_tasks(loop):
425 task.cancel()
426
427 for sig in (signal.SIGTERM, signal.SIGINT):
428 loop.add_signal_handler(sig, shutdown, sig)
429
430 # Run both tasks concurrently
431 try:
432 await asyncio.gather(
433 subscribe_sway(),
434 socket_server(),
435 )
436 except asyncio.CancelledError:
437 print("Shutdown complete", file=sys.stderr)
438 finally:
439 # Cleanup socket
440 if SOCKET_PATH.exists():
441 SOCKET_PATH.unlink()
442
443
444if __name__ == "__main__":
445 asyncio.run(main())