This repository has no description
17 kB
483 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# Subscribers: list of (StreamWriter, Event) tuples
275subscribers: list[tuple[asyncio.StreamWriter, asyncio.Event]] = []
276
277
278async def notify_subscribers(old_workspace: str | None, new_workspace: str) -> None:
279 """Notify all subscribers of a workspace change."""
280 event = (
281 json.dumps(
282 {
283 "event": "workspace_change",
284 "old": old_workspace,
285 "new": new_workspace,
286 }
287 ).encode()
288 + b"\n"
289 )
290 for writer, done in subscribers:
291 try:
292 writer.write(event)
293 await writer.drain()
294 except (OSError, ConnectionResetError):
295 done.set()
296
297
298async def subscribe_sway() -> None:
299 """Subscribe to sway IPC workspace events"""
300 print("Starting sway workspace subscription...", file=sys.stderr)
301
302 while True:
303 try:
304 proc = await asyncio.create_subprocess_exec(
305 "swaymsg",
306 "-t",
307 "subscribe",
308 "-m",
309 '["workspace"]',
310 stdout=asyncio.subprocess.PIPE,
311 stderr=asyncio.subprocess.PIPE,
312 )
313
314 if proc.stdout is None:
315 print("Failed to get stdout from swaymsg", file=sys.stderr)
316 await asyncio.sleep(5)
317 continue
318
319 async for line in proc.stdout:
320 try:
321 event: dict[str, Any] = json.loads(line.decode().strip())
322 if event.get("change") == "focus":
323 current: dict[str, Any] = event.get("current", {})
324 old: dict[str, Any] | None = event.get("old")
325 new_name: str | None = current.get("name")
326 old_name: str | None = old.get("name") if old else None
327
328 if new_name:
329 print(
330 f"Workspace change: {old_name} -> {new_name}",
331 file=sys.stderr,
332 )
333 await manager.handle_workspace_change(old_name, new_name)
334 await notify_subscribers(old_name, new_name)
335 except json.JSONDecodeError as e:
336 print(f"Failed to parse sway event: {e}", file=sys.stderr)
337
338 # If swaymsg exits, wait a bit and retry
339 await proc.wait()
340 print("swaymsg exited, retrying in 5 seconds...", file=sys.stderr)
341 await asyncio.sleep(5)
342
343 except FileNotFoundError:
344 print("swaymsg not found, retrying in 10 seconds...", file=sys.stderr)
345 await asyncio.sleep(10)
346 except Exception as e:
347 print(
348 f"Error in sway subscription: {e}, retrying in 5 seconds...",
349 file=sys.stderr,
350 )
351 await asyncio.sleep(5)
352
353
354async def handle_client(
355 reader: asyncio.StreamReader, writer: asyncio.StreamWriter
356) -> None:
357 """Handle a client connection to the Unix socket"""
358 try:
359 data = await reader.read(1024)
360 command = data.decode().strip()
361
362 response: dict[str, Any]
363 if command == "subscribe":
364 # Keep connection open and send workspace change events.
365 # Disconnection is detected in notify_subscribers when write fails.
366 done = asyncio.Event()
367 subscribers.append((writer, done))
368 print("New subscriber connected", file=sys.stderr)
369 try:
370 await done.wait()
371 except asyncio.CancelledError:
372 pass
373 finally:
374 subscribers[:] = [(w, d) for w, d in subscribers if w is not writer]
375 print("Subscriber disconnected", file=sys.stderr)
376 return
377 elif command == "dir":
378 response = {"directory": str(manager.get_current_directory())}
379 elif command == "workspace":
380 response = {"workspace": manager.current_workspace}
381 elif command == "status":
382 response = {
383 "workspace": manager.current_workspace,
384 "directory": str(manager.get_current_directory()),
385 "running": True,
386 }
387 else:
388 response = {"error": f"Unknown command: {command}"}
389
390 writer.write(json.dumps(response).encode() + b"\n")
391 await writer.drain()
392 except Exception as e:
393 print(f"Error handling client: {e}", file=sys.stderr)
394 finally:
395 writer.close()
396 await writer.wait_closed()
397
398
399async def socket_server() -> None:
400 """Run the Unix socket server for CLI queries"""
401 # Ensure socket directory exists
402 SOCKET_DIR.mkdir(parents=True, exist_ok=True)
403
404 # Remove stale socket if exists
405 if SOCKET_PATH.exists():
406 SOCKET_PATH.unlink()
407
408 server = await asyncio.start_unix_server(handle_client, path=str(SOCKET_PATH))
409
410 # Make socket accessible
411 os.chmod(SOCKET_PATH, 0o600)
412
413 print(f"Socket server listening on {SOCKET_PATH}", file=sys.stderr)
414
415 async with server:
416 await server.serve_forever()
417
418
419async def get_current_workspace() -> str | None:
420 """Get the current workspace from sway"""
421 try:
422 proc = await asyncio.create_subprocess_exec(
423 "swaymsg",
424 "-t",
425 "get_workspaces",
426 stdout=asyncio.subprocess.PIPE,
427 stderr=asyncio.subprocess.PIPE,
428 )
429 stdout, _ = await proc.communicate()
430 workspaces: list[dict[str, Any]] = json.loads(stdout.decode())
431 for ws in workspaces:
432 if ws.get("focused"):
433 name = ws.get("name")
434 return str(name) if name is not None else None
435 except Exception as e:
436 print(f"Failed to get current workspace: {e}", file=sys.stderr)
437 return None
438
439
440async def main() -> None:
441 """Main entry point"""
442 # Ensure user config directory exists (system config is read-only from nix store)
443 CONFIG_DIR.mkdir(parents=True, exist_ok=True)
444
445 # Log config directories
446 if SYSTEM_CONFIG_DIR:
447 print(f"System config dir: {SYSTEM_CONFIG_DIR}", file=sys.stderr)
448 print(f"User config dir: {CONFIG_DIR}", file=sys.stderr)
449
450 # Get initial workspace
451 initial_workspace = await get_current_workspace()
452 if initial_workspace:
453 manager.current_workspace = initial_workspace
454 manager._write_state()
455 print(f"Initial workspace: {initial_workspace}", file=sys.stderr)
456
457 # Handle shutdown gracefully
458 loop = asyncio.get_event_loop()
459
460 def shutdown(sig: signal.Signals) -> None:
461 print(f"Received {sig.name}, shutting down...", file=sys.stderr)
462 for task in asyncio.all_tasks(loop):
463 task.cancel()
464
465 for sig in (signal.SIGTERM, signal.SIGINT):
466 loop.add_signal_handler(sig, shutdown, sig)
467
468 # Run both tasks concurrently
469 try:
470 await asyncio.gather(
471 subscribe_sway(),
472 socket_server(),
473 )
474 except asyncio.CancelledError:
475 print("Shutdown complete", file=sys.stderr)
476 finally:
477 # Cleanup socket
478 if SOCKET_PATH.exists():
479 SOCKET_PATH.unlink()
480
481
482if __name__ == "__main__":
483 asyncio.run(main())