feat: add shell -r/--restart for race-free shell restarts

completions/caelestia.fish

@@ -27,6 +27,7 @@ complete -c caelestia -n $not_seen -a 'update' -d 'Update the Caelestia dotfiles
 set -l commands mpris drawers wallpaper notifs
 set -l not_seen "$seen shell && not $seen $commands"
 complete -c caelestia -n $not_seen -s 'd' -l 'daemon' -d 'Start the shell detached'
+complete -c caelestia -n $not_seen -s 'r' -l 'restart' -d 'Kill and restart the shell'
 complete -c caelestia -n $not_seen -s 's' -l 'show' -d 'Print all IPC commands'
 complete -c caelestia -n $not_seen -s 'l' -l 'log' -d 'Print the shell log'
 complete -c caelestia -n $not_seen -l 'log-rules' -d 'Log rules to apply'

src/caelestia/parser.py

@@ -37,6 +37,7 @@ def parse_args() -> tuple[argparse.ArgumentParser, argparse.Namespace]:
     shell_parser.set_defaults(cls=shell.Command)
     shell_parser.add_argument("message", nargs="*", help="a message to send to the shell")
     shell_parser.add_argument("-d", "--daemon", action="store_true", help="start the shell detached")
+    shell_parser.add_argument("-r", "--restart", action="store_true", help="kill and restart the shell")
     shell_parser.add_argument("-s", "--show", action="store_true", help="print all shell IPC commands")
     shell_parser.add_argument("-l", "--log", action="store_true", help="print the shell log")
     shell_parser.add_argument("-k", "--kill", action="store_true", help="kill the shell")

src/caelestia/subcommands/shell.py

@@ -1,6 +1,11 @@
+import json
+import os
+import signal
 import subprocess
+import time
 from argparse import Namespace
 
+from caelestia.utils.io import warn
 from caelestia.utils.paths import c_cache_dir
 
 
@@ -24,6 +29,11 @@ def run(self) -> None:
             # Send a message
             self.message(*self.args.message)
         else:
+            # Kill any running instance and wait for it to exit, otherwise `-n`
+            # will silently skip the relaunch
+            if self.args.restart:
+                self.stop_instances()
+
             # Start the shell
             args = ["qs", "-c", "caelestia", "-n"]
             if self.args.log_rules:
@@ -43,6 +53,43 @@ def run(self) -> None:
     def shell(self, *args: str) -> str:
         return subprocess.check_output(["qs", "-c", "caelestia", *args], text=True)
 
+    def list_instances(self) -> list[dict]:
+        proc = subprocess.run(["qs", "-c", "caelestia", "list", "-j"], capture_output=True, text=True)
+        try:
+            return json.loads(proc.stdout) if proc.returncode == 0 else []
+        except json.JSONDecodeError:
+            return []
+
+    def wait_for_exit(self, timeout: float) -> bool:
+        end = time.monotonic() + timeout
+        while time.monotonic() < end:
+            if not self.list_instances():
+                return True
+            time.sleep(0.1)
+        return False
+
+    def stop_instances(self) -> None:
+        instances = self.list_instances()
+        if not instances:
+            return
+
+        subprocess.run(["qs", "-c", "caelestia", "kill"], capture_output=True)
+
+        # Teardown is not instant (and slowest while a session lock is up)
+        if self.wait_for_exit(5):
+            return
+
+        # The instance is stuck; force kill it so the restart still happens
+        warn("shell did not exit gracefully, killing")
+        for instance in instances:
+            try:
+                os.kill(instance["pid"], signal.SIGKILL)
+            except (KeyError, ProcessLookupError):
+                pass
+
+        if not self.wait_for_exit(2):
+            warn("an instance of the shell is still running")
+
     def filter_log(self, line: str) -> bool:
         return f"Cannot open: file://{c_cache_dir}/imagecache/" not in line