fix(gateway): probe launchd domain instead of hardcoding user/<uid> (#40831)

The previous fix for #23387 changed _launchd_domain() from gui/<uid> to
user/<uid> to support Background/SSH sessions on macOS 26+. However, this
broke Aqua sessions where gui/<uid> is the only working domain and
user/<uid> cannot bootstrap or manage the service.

Now _launchd_domain() probes which domain actually contains the loaded
service:
1. Try gui/<uid> first (Aqua sessions)
2. Fall back to user/<uid> (Background/SSH sessions)
3. Use launchctl managername as heuristic when neither has the service
4. Cache the result for the process lifetime

Regression tests cover all four paths plus caching behavior.
This commit is contained in:
Tranquil-Flow 2026-06-07 02:09:04 +02:00 committed by Teknium
parent 2d75833abe
commit a8f404b29f
2 changed files with 196 additions and 8 deletions

View file

@ -495,7 +495,10 @@ class TestLaunchdServiceRecovery:
label = gateway_cli.get_launchd_label()
domain = gateway_cli._launchd_domain()
assert "--replace" in plist_path.read_text(encoding="utf-8")
assert calls[:2] == [
# The calls list includes launchctl print probes from _launchd_domain()
# before the bootout/bootstrap calls. Filter to only bootout/bootstrap.
service_calls = [c for c in calls if "bootout" in c or "bootstrap" in c]
assert service_calls[:2] == [
["launchctl", "bootout", f"{domain}/{label}"],
["launchctl", "bootstrap", domain, str(plist_path)],
]
@ -679,10 +682,22 @@ class TestLaunchdServiceRecovery:
assert "stale" in output.lower()
assert "not loaded" in output.lower()
def test_launchd_domain_uses_user_domain(self):
def test_launchd_domain_uses_user_domain(self, monkeypatch):
# The user/<uid> domain (not gui/<uid>) is the one reachable from
# non-Aqua/background sessions on macOS 26+ (issue #23387).
assert gateway_cli._launchd_domain() == f"user/{os.getuid()}"
# When gui/<uid> fails to probe and user/<uid> succeeds,
# _launchd_domain() must return user/<uid>.
gateway_cli._resolved_launchd_domain = None
monkeypatch.setattr(os, "getuid", lambda: 501)
label = gateway_cli.get_launchd_label()
def fake_run(cmd, check=False, **kwargs):
if "print" in cmd and "gui/" in " ".join(cmd):
raise subprocess.CalledProcessError(1, cmd, stderr="Domain error")
return SimpleNamespace(returncode=0, stdout="", stderr="")
monkeypatch.setattr(gateway_cli.subprocess, "run", fake_run)
assert gateway_cli._launchd_domain() == "user/501"
def test_launchctl_domain_unsupported_recognizes_macos26_codes(self):
# Codes that persist after a fresh bootstrap → launchd truly unavailable.
@ -836,6 +851,114 @@ class TestLaunchdServiceRecovery:
assert "nohup hermes gateway run" in out
class TestLaunchdDomainDetection:
"""Regression tests for _launchd_domain() probing (#40831).
The function must detect which launchd domain actually contains (or can
manage) the service, rather than hardcoding ``user/<uid>`` or ``gui/<uid>``.
"""
def _reset_domain_cache(self):
"""Clear any cached domain result between tests."""
gateway_cli._resolved_launchd_domain = None
def test_prefers_gui_domain_when_service_loaded_there(self, monkeypatch):
"""In an Aqua session where the service is loaded under gui/<uid>,
_launchd_domain() must return ``gui/<uid>`` not ``user/<uid>``."""
self._reset_domain_cache()
monkeypatch.setattr(os, "getuid", lambda: 501)
label = gateway_cli.get_launchd_label()
run_calls = []
def fake_run(cmd, check=False, **kwargs):
run_calls.append(cmd)
return SimpleNamespace(returncode=0, stdout="", stderr="")
monkeypatch.setattr(gateway_cli.subprocess, "run", fake_run)
domain = gateway_cli._launchd_domain()
assert domain == f"gui/501"
# Should have probed gui first
assert run_calls[0] == ["launchctl", "print", f"gui/501/{label}"]
def test_falls_back_to_user_domain_when_gui_fails(self, monkeypatch):
"""In a Background/SSH session where gui/<uid> fails but user/<uid>
works, _launchd_domain() must return ``user/<uid>``."""
self._reset_domain_cache()
monkeypatch.setattr(os, "getuid", lambda: 501)
label = gateway_cli.get_launchd_label()
run_calls = []
def fake_run(cmd, check=False, **kwargs):
run_calls.append(cmd)
if "print" in cmd and "gui/" in " ".join(cmd):
raise subprocess.CalledProcessError(1, cmd, stderr="Domain error")
return SimpleNamespace(returncode=0, stdout="", stderr="")
monkeypatch.setattr(gateway_cli.subprocess, "run", fake_run)
domain = gateway_cli._launchd_domain()
assert domain == f"user/501"
# Should have tried gui first, then user
assert len(run_calls) >= 2
def test_uses_managername_heuristic_when_both_probe_fail(self, monkeypatch):
"""When neither domain contains a loaded service, use
``launchctl managername`` as a tiebreaker: Aqua -> gui, else -> user."""
self._reset_domain_cache()
monkeypatch.setattr(os, "getuid", lambda: 501)
label = gateway_cli.get_launchd_label()
def fake_run(cmd, check=False, **kwargs):
if "print" in cmd:
raise subprocess.CalledProcessError(1, cmd, stderr="not found")
if "managername" in cmd:
return SimpleNamespace(returncode=0, stdout="Aqua\n", stderr="")
return SimpleNamespace(returncode=0, stdout="", stderr="")
monkeypatch.setattr(gateway_cli.subprocess, "run", fake_run)
domain = gateway_cli._launchd_domain()
assert domain == f"gui/501"
def test_managername_background_selects_user_domain(self, monkeypatch):
"""When managername is Background (non-Aqua), use user/<uid>."""
self._reset_domain_cache()
monkeypatch.setattr(os, "getuid", lambda: 501)
def fake_run(cmd, check=False, **kwargs):
if "print" in cmd:
raise subprocess.CalledProcessError(1, cmd, stderr="not found")
if "managername" in cmd:
return SimpleNamespace(returncode=0, stdout="Background\n", stderr="")
return SimpleNamespace(returncode=0, stdout="", stderr="")
monkeypatch.setattr(gateway_cli.subprocess, "run", fake_run)
domain = gateway_cli._launchd_domain()
assert domain == f"user/501"
def test_caches_result_across_calls(self, monkeypatch):
"""Domain detection should run once and cache the result."""
self._reset_domain_cache()
monkeypatch.setattr(os, "getuid", lambda: 501)
run_count = [0]
def fake_run(cmd, check=False, **kwargs):
run_count[0] += 1
return SimpleNamespace(returncode=0, stdout="", stderr="")
monkeypatch.setattr(gateway_cli.subprocess, "run", fake_run)
d1 = gateway_cli._launchd_domain()
d2 = gateway_cli._launchd_domain()
assert d1 == d2
assert run_count[0] == 1 # Only probed once
class TestGatewayServiceDetection:
def test_supports_systemd_services_requires_systemctl_binary(self, monkeypatch):
monkeypatch.setattr(gateway_cli, "is_linux", lambda: True)