refactor: simplify _get_service_pids — dedupe systemd scopes, fix self-import, harden launchd parsing

- Loop over user/system scope args instead of duplicating the systemd block
- Call get_launchd_label() directly instead of self-importing from hermes_cli.gateway
- Validate launchd output by checking parts[2] matches expected label (skip header)
- Add race-condition assumption docstring

hermes_cli/gateway.py

@@ -32,76 +32,53 @@ def _get_service_pids() -> set:
     """Return PIDs currently managed by systemd or launchd gateway services.
 
     Used to avoid killing freshly-restarted service processes when sweeping
-    for stale manual gateway processes after a service restart.
+    for stale manual gateway processes after a service restart.  Relies on the
+    service manager having committed the new PID before the restart command
+    returns (true for both systemd and launchd in practice).
     """
     pids: set = set()
 
-    # --- systemd (Linux) ---
+    # --- systemd (Linux): user and system scopes ---
     if is_linux():
-        try:
+        for scope_args in [["systemctl", "--user"], ["systemctl"]]:
-            result = subprocess.run(
+            try:
-                ["systemctl", "--user", "list-units", "hermes-gateway*",
+                result = subprocess.run(
-                 "--plain", "--no-legend", "--no-pager"],
+                    scope_args + ["list-units", "hermes-gateway*",
-                capture_output=True, text=True, timeout=5,
+                                  "--plain", "--no-legend", "--no-pager"],
-            )
+                    capture_output=True, text=True, timeout=5,
-            for line in result.stdout.strip().splitlines():
+                )
-                parts = line.split()
+                for line in result.stdout.strip().splitlines():
-                if not parts or not parts[0].endswith(".service"):
+                    parts = line.split()
-                    continue
+                    if not parts or not parts[0].endswith(".service"):
-                svc = parts[0]
+                        continue
-                try:
+                    svc = parts[0]
-                    show = subprocess.run(
+                    try:
-                        ["systemctl", "--user", "show", svc,
+                        show = subprocess.run(
-                         "--property=MainPID", "--value"],
+                            scope_args + ["show", svc,
-                        capture_output=True, text=True, timeout=5,
+                                          "--property=MainPID", "--value"],
-                    )
+                            capture_output=True, text=True, timeout=5,
-                    pid = int(show.stdout.strip())
+                        )
-                    if pid > 0:
+                        pid = int(show.stdout.strip())
-                        pids.add(pid)
+                        if pid > 0:
-                except (ValueError, subprocess.TimeoutExpired):
+                            pids.add(pid)
-                    pass
+                    except (ValueError, subprocess.TimeoutExpired):
-        except (FileNotFoundError, subprocess.TimeoutExpired):
+                        pass
-            pass
+            except (FileNotFoundError, subprocess.TimeoutExpired):
-
+                pass
-        # Also check system scope
-        try:
-            result = subprocess.run(
-                ["systemctl", "list-units", "hermes-gateway*",
-                 "--plain", "--no-legend", "--no-pager"],
-                capture_output=True, text=True, timeout=5,
-            )
-            for line in result.stdout.strip().splitlines():
-                parts = line.split()
-                if not parts or not parts[0].endswith(".service"):
-                    continue
-                svc = parts[0]
-                try:
-                    show = subprocess.run(
-                        ["systemctl", "show", svc,
-                         "--property=MainPID", "--value"],
-                        capture_output=True, text=True, timeout=5,
-                    )
-                    pid = int(show.stdout.strip())
-                    if pid > 0:
-                        pids.add(pid)
-                except (ValueError, subprocess.TimeoutExpired):
-                    pass
-        except (FileNotFoundError, subprocess.TimeoutExpired):
-            pass
 
     # --- launchd (macOS) ---
     if is_macos():
         try:
-            from hermes_cli.gateway import get_launchd_label
+            label = get_launchd_label()
             result = subprocess.run(
-                ["launchctl", "list", get_launchd_label()],
+                ["launchctl", "list", label],
                 capture_output=True, text=True, timeout=5,
             )
             if result.returncode == 0:
-                # Output format: "PID\tStatus\tLabel" header then data line
+                # Output: "PID\tStatus\tLabel" header, then one data line
                 for line in result.stdout.strip().splitlines():
                     parts = line.split()
-                    if parts:
+                    if len(parts) >= 3 and parts[2] == label:
                         try:
                             pid = int(parts[0])
                             if pid > 0:

tests/hermes_cli/test_update_gateway_restart.py

@@ -662,6 +662,7 @@ class TestGetServicePids:
     def test_returns_launchd_pid(self, monkeypatch):
         monkeypatch.setattr(gateway_cli, "is_linux", lambda: False)
         monkeypatch.setattr(gateway_cli, "is_macos", lambda: True)
+        monkeypatch.setattr(gateway_cli, "get_launchd_label", lambda: "ai.hermes.gateway")
 
         def fake_run(cmd, **kwargs):
             joined = " ".join(str(c) for c in cmd)