fix(terminal): serialize systemd scope capability probe

tests/tools/test_process_registry.py

@@ -1770,3 +1770,50 @@ class TestSystemdCgroupIsolation:
         assert first is True
         assert second is True
         assert len(probe_calls) == 1, "probe must run only once (cached)"
+
+    def test_systemd_scope_first_probe_is_serialized(self, monkeypatch):
+        """Concurrent first-use callers must wait for one definitive probe.
+
+        A temporary cached ``False`` would let a racing worker spawn inside the
+        gateway cgroup, defeating the OOM isolation guarantee.
+        """
+        import tools.process_registry as pr
+
+        monkeypatch.setattr(pr, "_SYSTEMD_SCOPE_AVAILABLE", None)
+        probe_started = threading.Event()
+        release_probe = threading.Event()
+        probe_calls = []
+        results = []
+
+        def fake_run(*args, **kwargs):
+            probe_calls.append(args)
+            probe_started.set()
+            assert release_probe.wait(timeout=2)
+            return subprocess.CompletedProcess(args=args[0], returncode=0)
+
+        monkeypatch.setattr("shutil.which", lambda name: "/usr/bin/systemd-run")
+        monkeypatch.setattr("subprocess.run", fake_run)
+
+        first = threading.Thread(
+            target=lambda: results.append(pr._systemd_run_user_scope_available())
+        )
+        second = threading.Thread(
+            target=lambda: results.append(pr._systemd_run_user_scope_available())
+        )
+        first.start()
+        assert probe_started.wait(timeout=2)
+        second.start()
+
+        # The racing caller must be blocked behind the probe, not observe a
+        # temporary False cache value.
+        second.join(timeout=0.05)
+        assert second.is_alive()
+
+        release_probe.set()
+        first.join(timeout=2)
+        second.join(timeout=2)
+
+        assert not first.is_alive()
+        assert not second.is_alive()
+        assert results == [True, True]
+        assert len(probe_calls) == 1

tools/process_registry.py

@@ -96,6 +96,7 @@ WATCH_GLOBAL_COOLDOWN_SECONDS = 30
 # services and containers), and cache the result for the process lifetime.
 
 _SYSTEMD_SCOPE_AVAILABLE: Optional[bool] = None
+_SYSTEMD_SCOPE_PROBE_LOCK = threading.Lock()
 
 
 def _systemd_run_user_scope_available() -> bool:
@@ -113,40 +114,49 @@ def _systemd_run_user_scope_available() -> bool:
     if _SYSTEMD_SCOPE_AVAILABLE is not None:
         return _SYSTEMD_SCOPE_AVAILABLE
 
-    _SYSTEMD_SCOPE_AVAILABLE = False
-    if _IS_WINDOWS:
-        return False
-    try:
-        import shutil
+    # 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
 
-        binary = shutil.which("systemd-run")
-        if not binary:
-            return False
-        # Probe: create a transient scope that immediately exits.  Use a
-        # unique unit name so concurrent probes don't collide.  A short
-        # timeout guards against a hung D-Bus.
-        probe_unit = f"hermes-probe-scope-{os.getpid()}-{int(time.time())}"
-        result = subprocess.run(
-            [
-                binary, "--user", "--scope", "--quiet",
-                "--unit", probe_unit,
-                "--collect", "--",
-                "/bin/true",
-            ],
-            capture_output=True,
-            timeout=3,
-        )
-        _SYSTEMD_SCOPE_AVAILABLE = result.returncode == 0
-        if not _SYSTEMD_SCOPE_AVAILABLE:
-            logger.debug(
-                "systemd-run --user --scope probe failed (rc=%s): %s",
-                result.returncode,
-                (result.stderr or b"").decode("utf-8", "replace").strip(),
-            )
-    except Exception as exc:
-        logger.debug("systemd-run --user --scope probe error: %s", exc)
-        _SYSTEMD_SCOPE_AVAILABLE = False
-    return _SYSTEMD_SCOPE_AVAILABLE
+        available = False
+        if not _IS_WINDOWS:
+            try:
+                import shutil
+
+                binary = shutil.which("systemd-run")
+                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())}"
+                    )
+                    result = subprocess.run(
+                        [
+                            binary, "--user", "--scope", "--quiet",
+                            "--unit", probe_unit,
+                            "--collect", "--",
+                            "/bin/true",
+                        ],
+                        capture_output=True,
+                        timeout=3,
+                    )
+                    available = result.returncode == 0
+                    if not available:
+                        logger.debug(
+                            "systemd-run --user --scope probe failed (rc=%s): %s",
+                            result.returncode,
+                            (result.stderr or b"").decode(
+                                "utf-8", "replace"
+                            ).strip(),
+                        )
+            except Exception as exc:
+                logger.debug("systemd-run --user --scope probe error: %s", exc)
+
+        _SYSTEMD_SCOPE_AVAILABLE = available
+        return available
 
 
 def _build_systemd_scope_argv(