This repository has no description
tools.workspace-manager: add sway workspace manager
- Python asyncio daemon subscribing to sway IPC events
- User configs in ~/workspaces/<name>.json for hooks and directories
- Declarative NixOS module with per-field config merging
- Shell hook for auto-cd in new terminals
- ruff/mypy checks integrated into package build
author
lassulus
date
(Feb 8, 2026, 5:35 PM CET)
commit
3700b7dd
3700b7ddedadbc4e6ee86ef9c59df8ff2a0642c2
parent
8a0407bf
8a0407bf1a0900567ea2bc6554caab0ab3b917b4
+594
+1
3modules/default.nix
+1
3modules/default.nix
+100
3modules/workspace-manager.nix
+100
3modules/workspace-manager.nix
···
1
+
{
2
+
config,
3
+
lib,
4
+
pkgs,
5
+
self,
6
+
...
7
+
}:
8
+
let
9
+
cfg = config.lass.workspace-manager;
10
+
11
+
workspaceType = lib.types.submodule {
12
+
options = {
13
+
directory = lib.mkOption {
14
+
type = lib.types.str;
15
+
description = "Directory to use for this workspace";
16
+
};
17
+
on_enter = lib.mkOption {
18
+
type = lib.types.listOf lib.types.str;
19
+
default = [ ];
20
+
description = "Commands to run when entering this workspace";
21
+
};
22
+
on_leave = lib.mkOption {
23
+
type = lib.types.listOf lib.types.str;
24
+
default = [ ];
25
+
description = "Commands to run when leaving this workspace";
26
+
};
27
+
};
28
+
};
29
+
30
+
# Generate a directory with JSON files for each declared workspace
31
+
systemConfigDir = pkgs.linkFarm "workspace-manager-configs" (
32
+
lib.mapAttrsToList (name: ws: {
33
+
name = "${name}.json";
34
+
path = pkgs.writeText "${name}.json" (
35
+
builtins.toJSON {
36
+
inherit (ws) directory on_enter on_leave;
37
+
}
38
+
);
39
+
}) cfg.workspaces
40
+
);
41
+
in
42
+
{
43
+
options.lass.workspace-manager = {
44
+
enable = lib.mkEnableOption "workspace manager daemon for sway";
45
+
46
+
package = lib.mkOption {
47
+
type = lib.types.package;
48
+
default = self.packages.${pkgs.system}.workspace-manager-daemon;
49
+
description = "The workspace-manager-daemon package to use";
50
+
};
51
+
52
+
workspaces = lib.mkOption {
53
+
type = lib.types.attrsOf workspaceType;
54
+
default = { };
55
+
description = "Declarative workspace configurations (take precedence over user configs)";
56
+
example = lib.literalExpression ''
57
+
{
58
+
dev = {
59
+
directory = "/home/user/src/myproject";
60
+
on_enter = [ "notify-send 'Entered dev workspace'" ];
61
+
};
62
+
music = {
63
+
directory = "/home/user/Music";
64
+
};
65
+
}
66
+
'';
67
+
};
68
+
69
+
configDir = lib.mkOption {
70
+
type = lib.types.str;
71
+
default = "~/workspaces";
72
+
description = "Directory containing user workspace configuration files";
73
+
};
74
+
75
+
defaultDirectory = lib.mkOption {
76
+
type = lib.types.str;
77
+
default = "~";
78
+
description = "Default directory for workspaces without configuration";
79
+
};
80
+
};
81
+
82
+
config = lib.mkIf cfg.enable {
83
+
systemd.user.services.workspace-manager = {
84
+
description = "Workspace Manager Daemon";
85
+
partOf = [ "sway-session.target" ];
86
+
wantedBy = [ "sway-session.target" ];
87
+
serviceConfig = {
88
+
Type = "simple";
89
+
ExecStart = "${cfg.package}/bin/workspace-manager-daemon";
90
+
Restart = "on-failure";
91
+
RestartSec = 5;
92
+
Environment = [
93
+
"WORKSPACE_MANAGER_CONFIG_DIR=${cfg.configDir}"
94
+
"WORKSPACE_MANAGER_SYSTEM_CONFIG_DIR=${systemConfigDir}"
95
+
"WORKSPACE_MANAGER_DEFAULT_DIR=${cfg.defaultDirectory}"
96
+
];
97
+
};
98
+
};
99
+
};
100
+
}
+349
tools/workspace-manager/daemon.py
+349
tools/workspace-manager/daemon.py
···
1
+
#!/usr/bin/env python3
2
+
"""
3
+
Workspace Manager Daemon
4
+
5
+
A 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
+
10
+
Config merging: user config is loaded first, then system config (declarative) overrides per-field
11
+
"""
12
+
13
+
from __future__ import annotations
14
+
15
+
import asyncio
16
+
import json
17
+
import os
18
+
import signal
19
+
import subprocess
20
+
import sys
21
+
from pathlib import Path
22
+
from typing import Any
23
+
24
+
# XDG runtime directory for socket and state
25
+
RUNTIME_DIR = Path(os.environ.get("XDG_RUNTIME_DIR", f"/run/user/{os.getuid()}"))
26
+
SOCKET_DIR = RUNTIME_DIR / "workspace-manager"
27
+
SOCKET_PATH = SOCKET_DIR / "socket"
28
+
STATE_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")
32
+
SYSTEM_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")
38
+
CONFIG_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")
45
+
DEFAULT_DIRECTORY = (
46
+
Path(_default_dir_env).expanduser() if _default_dir_env else Path.home()
47
+
)
48
+
49
+
50
+
def 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
+
62
+
class 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
+
103
+
class 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
+
176
+
manager = WorkspaceManager()
177
+
178
+
179
+
async 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
+
234
+
async 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
+
265
+
async 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
+
285
+
async 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
+
306
+
async 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
+
348
+
if __name__ == "__main__":
349
+
asyncio.run(main())
+47
tools/workspace-manager/flake-module.nix
+47
tools/workspace-manager/flake-module.nix
···
1
+
{ ... }:
2
+
{
3
+
perSystem =
4
+
{ pkgs, ... }:
5
+
let
6
+
python = pkgs.python3;
7
+
in
8
+
{
9
+
packages.workspace-manager = pkgs.writeShellApplication {
10
+
name = "workspace-manager";
11
+
runtimeInputs = [
12
+
pkgs.socat
13
+
pkgs.jq
14
+
];
15
+
text = builtins.readFile ./workspace-manager.sh;
16
+
};
17
+
18
+
packages.workspace-manager-daemon = pkgs.stdenv.mkDerivation {
19
+
pname = "workspace-manager-daemon";
20
+
version = "0.1.0";
21
+
src = ./.;
22
+
nativeBuildInputs = [
23
+
pkgs.makeWrapper
24
+
pkgs.ruff
25
+
python.pkgs.mypy
26
+
];
27
+
doCheck = true;
28
+
checkPhase = ''
29
+
runHook preCheck
30
+
ruff check daemon.py
31
+
ruff format --check daemon.py
32
+
mypy --strict daemon.py
33
+
runHook postCheck
34
+
'';
35
+
installPhase = ''
36
+
runHook preInstall
37
+
mkdir -p $out/bin
38
+
cp daemon.py $out/bin/workspace-manager-daemon
39
+
chmod +x $out/bin/workspace-manager-daemon
40
+
wrapProgram $out/bin/workspace-manager-daemon \
41
+
--prefix PATH : ${python}/bin:${pkgs.sway}/bin
42
+
runHook postInstall
43
+
'';
44
+
meta.mainProgram = "workspace-manager-daemon";
45
+
};
46
+
};
47
+
}
+90
tools/workspace-manager/workspace-manager.sh
+90
tools/workspace-manager/workspace-manager.sh
···
1
+
#!/usr/bin/env bash
2
+
# Workspace Manager CLI
3
+
# Queries the workspace-manager-daemon for current workspace and directory info
4
+
5
+
set -euo pipefail
6
+
7
+
RUNTIME_DIR="${XDG_RUNTIME_DIR:-/run/user/$(id -u)}"
8
+
SOCKET_PATH="$RUNTIME_DIR/workspace-manager/socket"
9
+
10
+
usage() {
11
+
echo "Usage: workspace-manager <command>"
12
+
echo ""
13
+
echo "Commands:"
14
+
echo " dir Print the directory for the current workspace"
15
+
echo " workspace Print the current workspace name"
16
+
echo " status Check if daemon is running and print status"
17
+
echo " help Show this help message"
18
+
exit 1
19
+
}
20
+
21
+
query_daemon() {
22
+
local command="$1"
23
+
if [[ ! -S "$SOCKET_PATH" ]]; then
24
+
echo "Error: Daemon not running (socket not found)" >&2
25
+
return 1
26
+
fi
27
+
28
+
# Send command and read response
29
+
response=$(echo "$command" | socat - "UNIX-CONNECT:$SOCKET_PATH" 2>/dev/null)
30
+
if [[ -z "$response" ]]; then
31
+
echo "Error: No response from daemon" >&2
32
+
return 1
33
+
fi
34
+
35
+
echo "$response"
36
+
}
37
+
38
+
cmd_dir() {
39
+
response=$(query_daemon "dir") || exit 1
40
+
echo "$response" | jq -r '.directory // empty'
41
+
}
42
+
43
+
cmd_workspace() {
44
+
response=$(query_daemon "workspace") || exit 1
45
+
echo "$response" | jq -r '.workspace // empty'
46
+
}
47
+
48
+
cmd_status() {
49
+
if [[ ! -S "$SOCKET_PATH" ]]; then
50
+
echo "Daemon is not running"
51
+
exit 1
52
+
fi
53
+
54
+
response=$(query_daemon "status") || exit 1
55
+
running=$(echo "$response" | jq -r '.running // false')
56
+
57
+
if [[ "$running" == "true" ]]; then
58
+
workspace=$(echo "$response" | jq -r '.workspace // "unknown"')
59
+
directory=$(echo "$response" | jq -r '.directory // "unknown"')
60
+
echo "Daemon is running"
61
+
echo " Workspace: $workspace"
62
+
echo " Directory: $directory"
63
+
else
64
+
echo "Daemon is not responding properly"
65
+
exit 1
66
+
fi
67
+
}
68
+
69
+
if [[ $# -lt 1 ]]; then
70
+
usage
71
+
fi
72
+
73
+
case "$1" in
74
+
dir)
75
+
cmd_dir
76
+
;;
77
+
workspace)
78
+
cmd_workspace
79
+
;;
80
+
status)
81
+
cmd_status
82
+
;;
83
+
help|--help|-h)
84
+
usage
85
+
;;
86
+
*)
87
+
echo "Unknown command: $1" >&2
88
+
usage
89
+
;;
90
+
esac
+7
tools/zsh/flake-module.nix
+7
tools/zsh/flake-module.nix
···
295
295
tmux set -g mouse on 2>/dev/null || true
296
296
fi
297
297
298
+
# Workspace manager integration: cd to workspace directory on new terminal
299
+
if [[ -z "$WORKSPACE_MANAGER_INIT" ]]; then
300
+
export WORKSPACE_MANAGER_INIT=1
301
+
_ws_dir=$(${self.packages.${pkgs.system}.workspace-manager}/bin/workspace-manager dir 2>/dev/null)
302
+
[[ -n "$_ws_dir" && -d "$_ws_dir" ]] && cd "$_ws_dir"
303
+
fi
304
+
298
305
# Disable some features to support TRAMP.
299
306
if [ "$TERM" = dumb ]; then
300
307
unsetopt zle prompt_cr prompt_subst