fix(tools): revalidate systemd scope availability

tests/tools/test_process_registry.py

@@ -2559,6 +2559,55 @@ class TestSystemdCgroupIsolation:
             value.startswith("OOMPolicy=") for value in probe_argv if isinstance(value, str)
         ), probe_argv
 
+    def test_successful_systemd_probe_uses_cached_verdict_before_ttl(self, monkeypatch):
+        """A healthy user bus does not cause a probe for every worker spawn."""
+        import tools.process_registry as pr
+
+        monkeypatch.setattr(pr, "_IS_LINUX", True)
+        monkeypatch.setattr(pr, "_SYSTEMD_SCOPE_AVAILABLE", None)
+        monkeypatch.setattr(pr, "_SYSTEMD_SCOPE_PROBED_AT", 0.0)
+        clock = [100.0]
+        probe_calls = []
+
+        def fake_run(*args, **kwargs):
+            probe_calls.append(args)
+            return subprocess.CompletedProcess(args=args[0], returncode=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 True
+        clock[0] += 30
+        assert pr._systemd_run_user_scope_available() is True
+        assert len(probe_calls) == 1
+
+    def test_successful_systemd_probe_revalidates_after_cache_ttl(self, monkeypatch):
+        """A vanished user bus invalidates a formerly successful scope verdict."""
+        import tools.process_registry as pr
+
+        monkeypatch.setattr(pr, "_IS_LINUX", True)
+        monkeypatch.setattr(pr, "_SYSTEMD_SCOPE_AVAILABLE", None)
+        monkeypatch.setattr(pr, "_SYSTEMD_SCOPE_PROBED_AT", 0.0)
+        clock = [100.0]
+        probe_results = [0, 1]
+        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 True
+        clock[0] += 61
+        assert pr._systemd_run_user_scope_available() is False
+        assert len(probe_calls) == 2
+
     @pytest.mark.linux_only
     def test_systemd_probe_derives_owned_user_bus_env_for_system_gateway(
         self, registry, monkeypatch, request

tools/process_registry.py

@@ -81,16 +81,17 @@ WATCH_GLOBAL_COOLDOWN_SECONDS = 30
 # Under a systemd gateway with MemoryMax, local background commands inherit the gateway's
 # cgroup, so a memory-heavy executor can get the ENTIRE gateway killed by systemd-oomd;
 # ``systemd-run --user --scope`` gives the worker its own transient cgroup. Usability is
-# probed once (binary present but user D-Bus absent in system services/containers).
+# probed with a bounded cache (binary present but user D-Bus absent in system services/containers).
 # A memory-heavy executor (Codex, tests, Node) can push the whole cgroup past MemoryMax and trigger
 # systemd-oomd to kill the ENTIRE gateway — taking down the messaging control plane and silently losing the
-# active turn. We probe *once* whether ``systemd-run --user --scope`` is actually usable (the binary can
+# active turn. We probe whether ``systemd-run --user --scope`` is actually usable (the binary can
 # exist on the PATH while the user D-Bus session is unavailable — common for system services and
-# containers), and cache the result for the process lifetime. See #70716.
+# containers), and cache the result briefly. See #70716.
 _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
+_SYSTEMD_SCOPE_SUCCESS_TTL_SECONDS = 60.0
 _MIN_WORKER_MEMORY_MAX_BYTES = 64 * 1024 * 1024
 _DEFAULT_WORKER_MEMORY_MAX_BYTES = 1024 * 1024 * 1024
 _WORKER_MEMORY_MAX_CAP_BYTES = 4 * 1024 * 1024 * 1024
@@ -204,12 +205,19 @@ def systemd_user_bus_env(base_env: Optional[Dict[str, str]] = None) -> Dict[str,
 
 
 def _systemd_scope_cached() -> Optional[bool]:
-    """Cached probe verdict, or None when a (re)probe is due. True is permanent; False
-    expires after ``_SYSTEMD_SCOPE_FAILURE_TTL_SECONDS`` so a D-Bus blip isn't sticky."""
-    if _SYSTEMD_SCOPE_AVAILABLE is True:
-        return True
-    stale = time.monotonic() - _SYSTEMD_SCOPE_PROBED_AT >= _SYSTEMD_SCOPE_FAILURE_TTL_SECONDS
-    return None if _SYSTEMD_SCOPE_AVAILABLE is None or stale else False
+    """Cached probe verdict, or None when a (re)probe is due.
+
+    Both verdicts expire: a user D-Bus can disappear after a successful probe,
+    while a failed probe can recover after linger or a login session starts.
+    """
+    if _SYSTEMD_SCOPE_AVAILABLE is None:
+        return None
+    ttl = (
+        _SYSTEMD_SCOPE_SUCCESS_TTL_SECONDS
+        if _SYSTEMD_SCOPE_AVAILABLE
+        else _SYSTEMD_SCOPE_FAILURE_TTL_SECONDS
+    )
+    return None if time.monotonic() - _SYSTEMD_SCOPE_PROBED_AT >= ttl else _SYSTEMD_SCOPE_AVAILABLE
 
 
 def _systemd_run_user_scope_available() -> bool: