fix(terminal): make systemd cleanup gateway-safe

tests/tools/test_process_registry.py

@@ -1717,6 +1717,71 @@ class TestSystemdCgroupIsolation:
         assert argv == ["/bin/bash", "-lic", "set +m; echo hello"], argv
         assert captured["start_new_session"] is True
 
+    def test_systemd_post_spawn_failure_never_kills_gateway_process_group(
+        self, registry, monkeypatch
+    ):
+        """The scope wrapper shares the gateway PG, so cleanup must not killpg."""
+        fake_popen, _captured = self._fake_popen_capture()
+        fake_proc = fake_popen(["placeholder"])
+
+        monkeypatch.setattr("tools.process_registry._find_shell", lambda: "/bin/bash")
+        monkeypatch.setattr(
+            "tools.process_registry._systemd_run_user_scope_available",
+            lambda: True,
+        )
+        monkeypatch.setattr(
+            "gateway.restart.is_gateway_supervisor_process",
+            lambda: True,
+        )
+        monkeypatch.setattr("shutil.which", lambda name: "/usr/bin/systemd-run")
+
+        broken_reader = MagicMock()
+        broken_reader.start.side_effect = RuntimeError("reader failed")
+
+        with patch("subprocess.Popen", return_value=fake_proc), \
+            patch("threading.Thread", return_value=broken_reader), \
+            patch("tools.process_registry._stop_systemd_unit", return_value=True) as stop_unit, \
+            patch("os.killpg") as killpg, \
+            patch.object(registry, "_write_checkpoint"):
+            with pytest.raises(RuntimeError, match="reader failed"):
+                registry.spawn_local("echo hello", cwd="/tmp")
+
+        stop_unit.assert_called_once()
+        assert stop_unit.call_args.args[0].startswith("hermes-worker-proc_")
+        assert stop_unit.call_args.args[0].endswith(".scope")
+        killpg.assert_not_called()
+
+    def test_pty_spawn_is_wrapped_in_systemd_scope(self, registry, monkeypatch):
+        """Interactive executors receive the same sibling-cgroup isolation."""
+        from ptyprocess import PtyProcess
+
+        fake_pty = MagicMock()
+        fake_pty.pid = 4321
+
+        monkeypatch.setattr("tools.process_registry._find_shell", lambda: "/bin/bash")
+        monkeypatch.setattr(
+            "tools.process_registry._systemd_run_user_scope_available",
+            lambda: True,
+        )
+        monkeypatch.setattr(
+            "gateway.restart.is_gateway_supervisor_process",
+            lambda: True,
+        )
+        monkeypatch.setattr("shutil.which", lambda name: "/usr/bin/systemd-run")
+
+        with patch.object(PtyProcess, "spawn", return_value=fake_pty) as pty_spawn, \
+            patch("threading.Thread", return_value=MagicMock()), \
+            patch.object(registry, "_write_checkpoint"):
+            session = registry.spawn_local("codex", cwd="/tmp", use_pty=True)
+
+        argv = pty_spawn.call_args.args[0]
+        assert argv[0] == "/usr/bin/systemd-run"
+        assert "--scope" in argv
+        assert "--unit" in argv
+        assert "--" in argv
+        assert argv[-3:] == ["/bin/bash", "-lic", "set +m; codex"]
+        assert session.systemd_unit == f"hermes-worker-{session.id}.scope"
+
     def test_kill_recovered_detached_already_exited_stops_persisted_scope(
         self, registry, monkeypatch
     ):
@@ -1817,3 +1882,45 @@ class TestSystemdCgroupIsolation:
         assert not second.is_alive()
         assert results == [True, True]
         assert len(probe_calls) == 1
+
+    def test_failed_systemd_probe_retries_after_cache_ttl(self, monkeypatch):
+        import tools.process_registry as pr
+
+        monkeypatch.setattr(pr, "_SYSTEMD_SCOPE_AVAILABLE", None)
+        monkeypatch.setattr(pr, "_SYSTEMD_SCOPE_PROBED_AT", 0.0, raising=False)
+        clock = [100.0]
+        probe_results = [1, 0]
+        probe_calls = []
+
+        def fake_run(*args, **kwargs):
+            probe_calls.append(args)
+            return subprocess.CompletedProcess(
+                args=args[0], returncode=probe_results.pop(0)
+            )
+
+        monkeypatch.setattr("shutil.which", lambda name: "/usr/bin/systemd-run")
+        monkeypatch.setattr("tools.process_registry.time.monotonic", lambda: clock[0])
+        monkeypatch.setattr("subprocess.run", fake_run)
+
+        assert pr._systemd_run_user_scope_available() is False
+        assert pr._systemd_run_user_scope_available() is False
+        assert len(probe_calls) == 1
+
+        clock[0] += 61
+        assert pr._systemd_run_user_scope_available() is True
+        assert len(probe_calls) == 2
+
+    def test_stop_systemd_unit_treats_absent_unit_as_clean(self, monkeypatch):
+        import tools.process_registry as pr
+
+        monkeypatch.setattr("shutil.which", lambda name: "/usr/bin/systemctl")
+        monkeypatch.setattr(
+            "subprocess.run",
+            lambda *args, **kwargs: subprocess.CompletedProcess(
+                args=args[0],
+                returncode=5,
+                stderr=b"Unit hermes-worker-gone.scope not loaded.\n",
+            ),
+        )
+
+        assert pr._stop_systemd_unit("hermes-worker-gone.scope") is True

tools/process_registry.py

@@ -97,6 +97,8 @@ WATCH_GLOBAL_COOLDOWN_SECONDS = 30
 
 _SYSTEMD_SCOPE_AVAILABLE: Optional[bool] = None
 _SYSTEMD_SCOPE_PROBE_LOCK = threading.Lock()
+_SYSTEMD_SCOPE_PROBED_AT = 0.0
+_SYSTEMD_SCOPE_FAILURE_TTL_SECONDS = 60.0
 
 
 def _systemd_run_user_scope_available() -> bool:
@@ -110,16 +112,31 @@ def _systemd_run_user_scope_available() -> bool:
     (``systemd-run --user --scope --unit=… -- /bin/true``) and remember the
     outcome.
     """
-    global _SYSTEMD_SCOPE_AVAILABLE
-    if _SYSTEMD_SCOPE_AVAILABLE is not None:
-        return _SYSTEMD_SCOPE_AVAILABLE
+    global _SYSTEMD_SCOPE_AVAILABLE, _SYSTEMD_SCOPE_PROBED_AT
+    cached = _SYSTEMD_SCOPE_AVAILABLE
+    now = time.monotonic()
+    if cached is True:
+        return True
+    if (
+        cached is False
+        and now - _SYSTEMD_SCOPE_PROBED_AT < _SYSTEMD_SCOPE_FAILURE_TTL_SECONDS
+    ):
+        return False
 
     # Double-checked locking keeps concurrent first-use spawns from observing
     # a temporary False while the definitive probe is still in flight.  Such a
     # race would launch the losing workload back inside the gateway cgroup.
     with _SYSTEMD_SCOPE_PROBE_LOCK:
-        if _SYSTEMD_SCOPE_AVAILABLE is not None:
-            return _SYSTEMD_SCOPE_AVAILABLE
+        cached = _SYSTEMD_SCOPE_AVAILABLE
+        now = time.monotonic()
+        if cached is True:
+            return True
+        if (
+            cached is False
+            and now - _SYSTEMD_SCOPE_PROBED_AT
+            < _SYSTEMD_SCOPE_FAILURE_TTL_SECONDS
+        ):
+            return False
 
         available = False
         if not _IS_WINDOWS:
@@ -130,9 +147,7 @@ def _systemd_run_user_scope_available() -> bool:
                 if binary:
                     # Probe: create a transient scope that immediately exits.
                     # A unique unit avoids collisions; timeout bounds D-Bus.
-                    probe_unit = (
-                        f"hermes-probe-scope-{os.getpid()}-{int(time.time())}"
-                    )
+                    probe_unit = f"hermes-probe-scope-{os.getpid()}-{uuid.uuid4().hex[:8]}"
                     result = subprocess.run(
                         [
                             binary, "--user", "--scope", "--quiet",
@@ -156,6 +171,7 @@ def _systemd_run_user_scope_available() -> bool:
                 logger.debug("systemd-run --user --scope probe error: %s", exc)
 
         _SYSTEMD_SCOPE_AVAILABLE = available
+        _SYSTEMD_SCOPE_PROBED_AT = time.monotonic()
         return available
 
 
@@ -215,10 +231,17 @@ def _stop_systemd_unit(unit_name: str) -> bool:
             timeout=15,
         )
         if result.returncode != 0:
+            stderr = (result.stderr or b"").decode(errors="replace").strip()
+            stderr_lower = stderr.lower()
+            if any(
+                marker in stderr_lower
+                for marker in ("not loaded", "not found", "does not exist")
+            ):
+                return True
             logger.debug(
                 "systemctl --user stop %s exited %d: %s",
                 unit_name, result.returncode,
-                result.stderr.decode(errors="replace").strip(),
+                stderr,
             )
             return False
         return True
@@ -969,7 +992,7 @@ class ProcessRegistry:
         # scope so it gets a separate cgroup.  An OOM in the worker then
         # kills only the worker instead of taking down the whole gateway
         # cgroup (and the messaging control plane with it).  We only do this
-        # for the common pipe-mode path; PTY mode is left as future work.
+        # for both pipe mode and the PTY path above.
         shell_argv = [user_shell, "-lic", f"set +m; {safe_command}"]
         use_systemd_scope = False
         under_supervisor = False
@@ -1048,7 +1071,14 @@ class ProcessRegistry:
             # descendants spawned via setsid) before re-raising so they do not
             # leak as untracked background processes.
             try:
-                if not _IS_WINDOWS:
+                if session.systemd_unit:
+                    # systemd-run --scope shares the gateway's process group
+                    # because start_new_session=False.  Never killpg here: that
+                    # can signal the gateway itself.  Stop the sibling cgroup,
+                    # then safely terminate the wrapper PID tree as fallback.
+                    _stop_systemd_unit(session.systemd_unit)
+                    self._terminate_host_pid(proc.pid, session.host_start_time)
+                elif not _IS_WINDOWS:
                     try:
                         kill_signal = getattr(signal, "SIGKILL", signal.SIGTERM)
                         os.killpg(os.getpgid(proc.pid), kill_signal)  # windows-footgun: ok - guarded by _IS_WINDOWS above