hermes-admin/hermes-agent
1
0
Fork 0
Code Issues Pull Requests Actions 33 Packages Projects Releases Wiki Activity

test: restore gateway lifecycle and doctor TCC guards dropped by #120071

Browse Source 
Restored (OS fakes replaced by real windows_only/macos_only markers):
- test_doctor.py: macOS TCC grant check — silent off macOS / without bundle, warns on cdhash-pinned DR, stable for identifier and certificate-anchored DRs, never claims stable when codesign output is unreadable (#86385, #73681).
- test_gateway.py: Windows stop drains before force-kill, force-kills only after drain timeout, kill carries pre-drain start time so a recycled PID is refused (#112750); orphan reaper spares the launchd gateway (#74075), the recorded PID and Scheduled Task bootstrap parent (#86098), even when validation fails (#87158).
- test_gateway_proc_fallback.py: ps fallback uses BSD-compatible flags/columns (#74075); launchd PID collection for current and all profiles (#73627).
- test_windows_gateway_cold_start_desktop_lifecycle.py: dead attested profiles cold-start beside a running one / when nothing runs; service-supervised running profile is not cold-started (#110959).
This commit is contained in:
teknium1
2026-09-23 04:39:18 -07:00
committed by Teknium
parent 76cfc7b257
commit b5da2f594f
4 changed files with 563 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

70
tests/hermes_cli/test_doctor.py
View File

@@ -1373,12 +1373,82 @@ class TestDoctorDeprecatedConfigAndEnv:
assert doctor_config.collect_deprecated_env_vars(None) == []
@pytest.mark.linux_only
def test_macos_tcc_grant_check_is_silent_off_macos(monkeypatch, capsys, tmp_path):
"""Off macOS the TCC check prints nothing, even with a bundle present."""
monkeypatch.setattr(doctor_platform, "_desktop_app_bundle", lambda: tmp_path / "Hermes.app")
doctor_platform.check_macos_tcc_grants()
assert capsys.readouterr().out == ""
@pytest.mark.macos_only
class TestMacOSTCCGrants:
"""macOS TCC grant persistence check (#86385): a cdhash-pinned DR (pre-#73681
local builds) silently resets Screen Recording/Accessibility grants on every
rebuild while the Settings toggle stays ON."""
@staticmethod
def _darwin_bundle(monkeypatch, tmp_path, dr):
monkeypatch.setattr(doctor_platform, "_desktop_app_bundle", lambda: tmp_path / "Hermes.app")
if dr is not ...:
monkeypatch.setattr(doctor_platform, "_macos_desktop_dr", lambda app: dr)
def test_silent_without_desktop_bundle(self, monkeypatch, capsys):
monkeypatch.setattr(doctor_platform, "_desktop_app_bundle", lambda: None)
doctor_platform.check_macos_tcc_grants()
assert capsys.readouterr().out == ""
def test_warns_on_cdhash_pinned_dr(self, monkeypatch, capsys, tmp_path):
self._darwin_bundle(
monkeypatch, tmp_path,
'designated => identifier "com.nousresearch.hermes" and cdhash H"97e692f3890f781fa0ad5ad6cb9d769cfaf42628"',
)
doctor_platform.check_macos_tcc_grants()
out = capsys.readouterr().out
assert "TCC grants will reset after every update" in out
assert "hermes update" in out
assert "signing identity is stable" not in out
def test_identifier_dr_is_stable_with_upgrade_hint_and_repair_info(self, monkeypatch, capsys, tmp_path):
self._darwin_bundle(monkeypatch, tmp_path, 'designated => identifier "com.nousresearch.hermes"')
doctor_platform.check_macos_tcc_grants()
out = capsys.readouterr().out
assert "TCC signing identity is stable" in out
assert "--setup-tcc-identity" in out
assert "tccutil reset ScreenCapture com.nousresearch.hermes" in out
def test_certificate_anchored_dr_is_stable_without_upgrade_hint(self, monkeypatch, capsys, tmp_path):
self._darwin_bundle(
monkeypatch, tmp_path,
'designated => identifier "com.nousresearch.hermes" and certificate root = H"aabbcc"',
)
doctor_platform.check_macos_tcc_grants()
out = capsys.readouterr().out
assert "TCC signing identity is stable" in out
assert "--setup-tcc-identity" not in out
assert "tccutil reset ScreenCapture com.nousresearch.hermes" in out
@pytest.mark.parametrize("failure", ["none", "empty", "timeout", "no_codesign"])
def test_unreadable_dr_warns_and_never_claims_stable(self, monkeypatch, capsys, tmp_path, failure):
"""codesign failing, hanging, missing or printing nothing degrades to a
warning; an empty DR must not false-positive as a stable identity."""
if failure in ("none", "empty"):
self._darwin_bundle(monkeypatch, tmp_path, None if failure == "none" else "")
else:
self._darwin_bundle(monkeypatch, tmp_path, ...)
if failure == "timeout":
monkeypatch.setattr(shutil, "which", lambda _name: "/usr/bin/codesign")
def _timeout(*args, **kwargs):
raise subprocess.TimeoutExpired(cmd=["codesign"], timeout=15)
monkeypatch.setattr(subprocess, "run", _timeout)
else:
monkeypatch.setattr(shutil, "which", lambda _name: None)
doctor_platform.check_macos_tcc_grants()
out = capsys.readouterr().out
assert "could not read code-signing requirement" in out
assert "stable" not in out
def test_run_doctor_reports_shadowed_lightpanda_engine(monkeypatch, tmp_path):

264
tests/hermes_cli/test_gateway.py
View File

@@ -503,7 +503,98 @@ class TestRestartWaitsForApiServerPort:
class TestStopProfileGateway:
@pytest.mark.windows_only
def test_windows_stop_drains_marker_before_force_termination(self, monkeypatch):
"""Windows must let the marker watcher run before escalating (#112750)."""
import hermes_cli.gateway_windows as gateway_windows
pid = 12345
calls = []
monkeypatch.setattr("gateway.status.get_running_pid", lambda: pid)
monkeypatch.setattr(gateway_windows, "_windows_stop_drain_timeout", lambda: 7.0)
monkeypatch.setattr(
gateway_windows,
"_drain_gateway_pid",
lambda target, timeout: calls.append(("drain", target, timeout)) or True,
)
monkeypatch.setattr(
gateway_windows,
"_force_terminate_known_gateway_pids",
lambda pids: calls.append(("force", pids)),
)
monkeypatch.setattr(gateway.os, "kill", lambda *_: calls.append(("kill",)))
monkeypatch.setattr("gateway.status._pid_exists", lambda target: False)
monkeypatch.setattr("gateway.status.remove_pid_file", lambda: None)
monkeypatch.setattr(gateway, "_reap_unsupervised_gateway_orphans", lambda **_: False)
assert gateway.stop_profile_gateway() is True
assert calls == [("drain", pid, 7.0)]
@pytest.mark.windows_only
def test_windows_stop_force_terminates_only_after_drain_timeout(self, monkeypatch):
"""A wedged Windows gateway still has a bounded force-stop fallback (#112750)."""
import hermes_cli.gateway_windows as gateway_windows
pid = 12345
calls = []
monkeypatch.setattr("gateway.status.get_running_pid", lambda: pid)
monkeypatch.setattr(gateway_windows, "_windows_stop_drain_timeout", lambda: 7.0)
monkeypatch.setattr("gateway.status.get_process_start_time", lambda target: 100)
monkeypatch.setattr(
gateway_windows,
"_drain_gateway_pid",
lambda target, timeout: calls.append(("drain", target, timeout)) or False,
)
monkeypatch.setattr(
gateway_windows,
"_force_terminate_known_gateway_pids",
lambda pids: calls.append(("force", pids)),
)
monkeypatch.setattr(gateway.os, "kill", lambda *_: calls.append(("kill",)))
monkeypatch.setattr("gateway.status._pid_exists", lambda target: False)
monkeypatch.setattr("gateway.status.remove_pid_file", lambda: None)
monkeypatch.setattr(gateway, "_reap_unsupervised_gateway_orphans", lambda **_: False)
assert gateway.stop_profile_gateway() is True
assert calls == [("drain", pid, 7.0), ("force", {pid: 100})]
@pytest.mark.windows_only
def test_windows_stop_force_kill_carries_pre_drain_identity(self, monkeypatch):
"""The post-drain taskkill must be guarded by the start time captured BEFORE the <=30 s drain:
a PID recycled during the wait shows a different start time, so ``terminate_pid``'s mismatch
refusal fires instead of killing an unrelated process. Reading it at kill time is a vacuous
self-comparison."""
import gateway.status as status
import hermes_cli.gateway_windows as gateway_windows
pid = 4242
calls = []
clock = {"now": 0.0}
alive = {"value": True}
monkeypatch.setattr(gateway_windows.time, "monotonic", lambda: clock["now"])
monkeypatch.setattr(
gateway_windows.time, "sleep", lambda _s: clock.__setitem__("now", clock["now"] + 10.0)
)
monkeypatch.setattr(gateway_windows, "_windows_stop_drain_timeout", lambda: 7.0)
monkeypatch.setattr(status, "get_running_pid", lambda: pid if alive["value"] else None)
monkeypatch.setattr(status, "write_planned_stop_marker", lambda target: None)
monkeypatch.setattr(status, "_pid_exists", lambda target: alive["value"])
# The original gateway (start time 100) exits during the drain and its PID is recycled (999).
monkeypatch.setattr(
status, "get_process_start_time", lambda target: 100 if clock["now"] < 7.0 else 999
)
def _terminate(target, force=False, expected_start_time=None):
calls.append((target, force, expected_start_time))
alive["value"] = False
monkeypatch.setattr(status, "terminate_pid", _terminate)
monkeypatch.setattr(status, "remove_pid_file", lambda: None)
monkeypatch.setattr(gateway, "_reap_unsupervised_gateway_orphans", lambda **_: False)
assert gateway.stop_profile_gateway() is True
assert calls == [(pid, True, 100)]
def test_stop_profile_gateway_keeps_pid_file_when_process_still_running(self, monkeypatch):
calls = {"kill": 0, "alive_probes": 0, "remove": 0, "reap_calls": 0}
@@ -563,6 +654,179 @@ class TestStopProfileGateway:
assert killed_pid in reap_extra_excludes[0]
@pytest.mark.macos_only
class TestReapUnsupervisedGatewayOrphansMacOS:
"""Tests that the orphan reaper excludes launchd-managed PIDs on macOS.
Regression guard: without the ``is_macos()`` exclusion of
``_get_service_pids()``, the reaper would SIGTERM the launchd-supervised
gateway every time Hermes Desktop opens (``hermes serve`` calls
``_reap_unsupervised_gateway_orphans`` during startup).
"""
def test_macos_excludes_launchd_pid_from_kill(self, monkeypatch):
"""A launchd-managed PID must not appear in the orphan kill list."""
launchd_pid = 52615
monkeypatch.setattr(gateway, "supports_systemd_services", lambda: False)
# _get_service_pids returns the launchd-managed gateway PID.
# (accepts all_profiles: the reaper asks for the whole fleet, #74075)
monkeypatch.setattr(
gateway, "_get_service_pids", lambda all_profiles=False: {launchd_pid}
)
# No pidfile-recorded gateway in this scenario.
monkeypatch.setattr("gateway.status.get_running_pid", lambda: None)
# find_gateway_pids returns the launchd PID plus a real orphan.
# The reaper should only kill the orphan, not the launchd PID.
orphan_pid = 99998
monkeypatch.setattr(
gateway,
"find_gateway_pids",
lambda exclude_pids=None: [p for p in [launchd_pid, orphan_pid] if p not in (exclude_pids or set())],
)
killed_pids = []
monkeypatch.setattr(gateway.os, "kill", lambda pid, sig: killed_pids.append((pid, sig)))
monkeypatch.setattr("gateway.status._pid_exists", lambda pid: False)
monkeypatch.setattr("gateway.status.write_planned_stop_marker", lambda pid: None)
monkeypatch.setattr("time.sleep", lambda _: None)
monkeypatch.setattr("time.monotonic", lambda: 1.0)
result = gateway._reap_unsupervised_gateway_orphans()
assert result is True # at least one orphan was reaped
killed = [pid for pid, _ in killed_pids]
assert orphan_pid in killed # the real orphan was killed
assert launchd_pid not in killed # the launchd PID was NOT killed
@pytest.mark.windows_only
class TestReapUnsupervisedGatewayOrphansWindows:
"""Tests that the orphan reaper spares the recorded gateway PID and its
supervision chain on Windows.
Regression guard: without the Windows exemption of the recorded healthy
gateway PID (and its parent chain), the reaper would SIGTERM/SIGKILL a
Scheduled-Task-supervised gateway every time Hermes Desktop opens
(``hermes serve`` calls ``_reap_unsupervised_gateway_orphans`` during
startup). The Scheduled-Task bootstrap's argv matches the gateway scan,
so it is reaped as an "orphan" — and when the bootstrap dies, the
detached gateway it spawned exits with it (#86098).
"""
@staticmethod
def _install_fake_psutil(monkeypatch, chain):
"""Install a fake psutil module exposing the given process chain."""
by_pid = {proc.pid: proc for proc in chain}
fake_psutil = SimpleNamespace(Process=lambda pid: by_pid[pid])
monkeypatch.setitem(sys.modules, "psutil", fake_psutil)
def test_windows_excludes_recorded_pid_and_bootstrap_from_kill(self, monkeypatch):
"""The recorded gateway PID and its bootstrap parent must not be killed."""
recorded_pid = 52615 # detached gateway recorded in gateway.pid
bootstrap_pid = 52616 # Scheduled-Task bootstrap (argv matches scan)
orphan_pid = 99998 # a real orphan that should still be reaped
monkeypatch.setattr(gateway, "supports_systemd_services", lambda: False)
# gateway.pid records the detached gateway; its parent is the
# Scheduled-Task bootstrap whose argv matches the gateway scan.
bootstrap = SimpleNamespace(pid=bootstrap_pid, parent=lambda: None)
recorded = SimpleNamespace(pid=recorded_pid, parent=lambda: bootstrap)
self._install_fake_psutil(monkeypatch, [recorded, bootstrap])
# get_running_pid() returns the recorded healthy gateway PID.
monkeypatch.setattr(
"gateway.status.get_running_pid", lambda cleanup_stale=True: recorded_pid
)
# find_gateway_pids returns the recorded PID, its bootstrap parent
# and a real orphan. The reaper should only kill the orphan.
monkeypatch.setattr(
gateway,
"find_gateway_pids",
lambda exclude_pids=None: [
p
for p in [recorded_pid, bootstrap_pid, orphan_pid]
if p not in (exclude_pids or set())
],
)
killed_pids = []
marked_pids = []
monkeypatch.setattr(gateway.os, "kill", lambda pid, sig: killed_pids.append((pid, sig)))
monkeypatch.setattr("gateway.status._pid_exists", lambda pid: False)
monkeypatch.setattr("gateway.status.write_planned_stop_marker", marked_pids.append)
monkeypatch.setattr("time.sleep", lambda _: None)
monkeypatch.setattr("time.monotonic", lambda: 1.0)
result = gateway._reap_unsupervised_gateway_orphans()
assert result is True # at least one orphan was reaped
assert marked_pids == [orphan_pid] # the real orphan was asked to drain
assert killed_pids == [] # Windows SIGTERM must not TerminateProcess
def test_windows_raw_record_supplies_exclusion_when_validation_fails(
self, monkeypatch
):
"""A registration that fails liveness VALIDATION must still shield
the recorded PID from the sweep.
get_running_pid returns None whenever the record fails validation
(start-time mismatch, argv drift, lock hiccup) — regardless of
cleanup_stale, which only controls unlinking. The exclusion set is
therefore built from the RAW pidfile/lock records: a stale recorded
PID at worst spares one process, while a validation false-negative
would TerminateProcess a healthy standalone gateway (#87158).
"""
recorded_pid = 52615
monkeypatch.setattr(gateway, "supports_systemd_services", lambda: False)
recorded = SimpleNamespace(pid=recorded_pid, parent=lambda: None)
self._install_fake_psutil(monkeypatch, [recorded])
# The raw record is present on disk; the validated probe rejects it.
monkeypatch.setattr(
"gateway.status._read_pid_record", lambda *a, **k: {"pid": recorded_pid}
)
monkeypatch.setattr(
"gateway.status._read_gateway_lock_record", lambda *a, **k: None
)
probe_kwargs = []
def failing_validation(*a, cleanup_stale=True, **k):
probe_kwargs.append(cleanup_stale)
return None
monkeypatch.setattr(
"gateway.status.get_running_pid", failing_validation
)
monkeypatch.setattr(
gateway,
"find_gateway_pids",
lambda exclude_pids=None: [
p for p in [recorded_pid] if p not in (exclude_pids or set())
],
)
killed_pids = []
monkeypatch.setattr(
gateway.os, "kill", lambda pid, sig: killed_pids.append((pid, sig))
)
result = gateway._reap_unsupervised_gateway_orphans()
assert probe_kwargs == [False], (
"the fallback probe must still be non-destructive so the sweep "
f"never unlinks the record it just read, got {probe_kwargs}"
)
assert result is False
assert killed_pids == [] # the standalone gateway survived
class TestReaperCandidateIsSupervisorOwned:
"""Regression for the Windows pidfile-less supervisor-owned case (#83683).

112
tests/hermes_cli/test_gateway_proc_fallback.py
View File

@@ -114,14 +114,126 @@ class TestProcFallback:
mock_ps.assert_not_called() # /proc dir existed, so ps not called
class TestPsFallbackBsdCompat:
"""The ps fallback must use flags BSD/macOS ps accepts (#73626, #74075).
``ps -A eww`` fails on macOS (BSD ``e`` is not the procps flag), which
made gateway discovery silently return nothing whenever /proc is absent.
"""
def test_ps_fallback_uses_bsd_compatible_flags_and_columns(self):
with (
patch("hermes_cli.gateway.is_windows", return_value=False),
patch("os.path.isdir", side_effect=lambda p: p != "/proc"),
patch("hermes_cli.gateway._get_ancestor_pids", return_value=set()),
patch("subprocess.run") as mock_run,
):
mock_run.return_value = MagicMock(returncode=1, stdout="", stderr="")
assert not gateway_mod._scan_gateway_pids(set())
ps_calls = [
c[0][0] for c in mock_run.call_args_list if c[0] and c[0][0] and c[0][0][0] == "ps"
]
assert ps_calls, "ps was not invoked at all"
ps_call = ps_calls[0]
assert "-Aww" in ps_call and "eww" not in " ".join(ps_call), ps_call
assert "-o" in ps_call and "pid=,command=" in ps_call, ps_call
class TestGetServicePidsAllProfiles:
"""_get_service_pids(all_profiles=...) discovery across profiles."""
@pytest.mark.macos_only
def test_default_scope_uses_current_profile_label(self):
"""Without all_profiles, only the current profile's launchd agent is
located (per-label domain-explicit probe, #73627)."""
located = []
def _fake_locate(label):
located.append(label)
return ("gui/501", 123)
with (
patch("hermes_cli.gateway.supports_systemd_services", return_value=False),
patch(
"hermes_cli.gateway.get_launchd_label",
return_value="ai.hermes.gateway.myprofile",
),
patch(
"hermes_cli.gateway._locate_launchd_gateway_service",
side_effect=_fake_locate,
),
patch("subprocess.run") as mock_run,
):
pids = gateway_mod._get_service_pids()
assert pids == {123}
# Default scope: exactly the current profile's label, no fleet
# enumeration and no bare `launchctl list` scan.
assert located == ["ai.hermes.gateway.myprofile"]
launchctl_calls = [
c[0][0]
for c in mock_run.call_args_list
if c[0] and c[0][0] and c[0][0][0] == "launchctl"
]
assert launchctl_calls == []
@pytest.mark.macos_only
def test_all_profiles_enumerates_all_gateway_labels(self):
"""With all_profiles=True, every install-derived gateway label is
located (#73627), and the bare ``launchctl list`` prefix scan still
widens the EXCLUDE set with unmapped ai.hermes.gateway* agents
(#74075 belt-and-suspenders)."""
located = []
label_pids = {
"ai.hermes.gateway": 123,
"ai.hermes.gateway-profile-b": 456,
}
def _fake_locate(label):
located.append(label)
pid = label_pids.get(label)
return ("gui/501", pid) if pid else (None, None)
with (
patch("hermes_cli.gateway.supports_systemd_services", return_value=False),
patch(
"hermes_cli.gateway.get_launchd_label",
return_value="ai.hermes.gateway",
),
patch(
"hermes_cli.gateway.launchd_gateway_labels_for_install",
return_value=["ai.hermes.gateway", "ai.hermes.gateway-profile-b"],
),
patch(
"hermes_cli.gateway._locate_launchd_gateway_service",
side_effect=_fake_locate,
),
patch("subprocess.run") as mock_run,
):
mock_run.return_value = MagicMock(
returncode=0,
stdout=(
"999\t0\tai.hermes.gateway-unmapped\n"
"789\t0\tcom.apple.some.other.agent\n"
),
stderr="",
)
pids = gateway_mod._get_service_pids(all_profiles=True)
# Label-derived fleet + prefix-scan stragglers; non-gateway excluded.
assert pids == {123, 456, 999}
assert 789 not in pids
assert sorted(located) == [
"ai.hermes.gateway",
"ai.hermes.gateway-profile-b",
]
launchctl_calls = [
c[0][0]
for c in mock_run.call_args_list
if c[0] and c[0][0] and c[0][0][0] == "launchctl"
]
assert launchctl_calls == [["launchctl", "list"]]
def test_all_profiles_preserves_systemd_behavior(self):
"""systemd scope is unaffected by the all_profiles switch — it already

117
tests/hermes_cli/test_windows_gateway_cold_start_desktop_lifecycle.py
View File

@@ -18,10 +18,16 @@ does not restart the messaging gateway itself.
from __future__ import annotations
import json
import pytest
from hermes_cli import gateway as hermes_gateway
from hermes_cli import gateway_windows
from hermes_cli import main as cli_main
from hermes_cli import process_identity
from hermes_cli import update_cmd
import hermes_cli.update_cmd_windows as update_cmd_windows
def _live_serve_ledger_entry() -> dict:
@@ -90,3 +96,114 @@ def test_orphaned_control_plane_does_not_own_lifecycle(monkeypatch):
assert update_cmd._desktop_owns_gateway_lifecycle() is False
def _running_beta_pause_fixture(monkeypatch, tmp_path):
"""Windows update with ``beta`` (PID 777) running and the default profile home at ``tmp_path``."""
from types import SimpleNamespace
import hermes_cli.profiles as profiles_mod
homes = {"default": tmp_path, "beta": tmp_path / "profiles" / "beta"}
homes["beta"].mkdir(parents=True)
monkeypatch.setattr("hermes_cli.config.get_hermes_home", lambda: str(tmp_path))
monkeypatch.setattr(update_cmd, "_desktop_owns_gateway_lifecycle", lambda: True)
monkeypatch.setattr(update_cmd_windows, "_desktop_owns_gateway_lifecycle", lambda: True)
beta = SimpleNamespace(pid=777, profile="beta")
monkeypatch.setattr(update_cmd_windows, "_discover_windows_gateways", lambda: ({777: beta}, [], set(), [777]))
monkeypatch.setattr(update_cmd_windows, "_request_socket_pauses", lambda *a: ({"beta": 777}, [777], []))
monkeypatch.setattr(cli_main, "_venv_launcher_ancestors", lambda pids: [])
monkeypatch.setattr(cli_main, "_wait_for_windows_update_gateway_exit", lambda pids, timeout: set())
monkeypatch.setattr(profiles_mod, "get_active_profile_name", lambda: "default")
monkeypatch.setattr(profiles_mod, "profiles_to_serve", lambda multiplex, **_kw: [(n, h) for n, h in homes.items() if n in ("default", "beta")])
monkeypatch.setattr(profiles_mod, "get_profile_dir", lambda name: homes[name])
# Resume side.
monkeypatch.setattr(cli_main, "_refresh_windows_gateway_launchers", lambda: None)
monkeypatch.setattr(hermes_gateway, "launch_detached_profile_gateway_restart", lambda p, o: True)
ready_probes: list = []
monkeypatch.setattr(gateway_windows, "_wait_for_gateway_ready", lambda *a, **k: ready_probes.append(k) or [4242])
homes["_ready_probes"] = ready_probes
return homes
@pytest.mark.windows_only
def test_dead_attested_default_is_cold_started_beside_running_beta(monkeypatch, tmp_path, capsys):
"""#110959: the all-or-nothing plan never ran while ``beta`` was alive, so a default gateway
that died after a ✓ stayed down after the update. Its dead attestation must become a per-profile
cold-start obligation on the token; resume spawns it under the default home and consumes only
that generation, while ``beta`` still goes through the ordinary relaunch."""
homes = _running_beta_pause_fixture(monkeypatch, tmp_path)
gateway_windows._write_start_attestation([555], "direct spawn (PID 555)")
marker = tmp_path / "state" / "gateway.start-attestation.json"
generation = json.loads(marker.read_text(encoding="utf-8"))["generation"]
token = update_cmd._pause_windows_gateways_for_update()
assert token["profiles"] == {"beta": 777}
assert token["cold_start_profiles"] == {"default": generation}
assert marker.exists() # plan-time probe is read-only
spawned = []
monkeypatch.setattr(gateway_windows, "_spawn_detached", lambda **k: spawned.append(k) or 4242)
update_cmd._resume_windows_gateways_after_update(token)
assert spawned == [{"home": homes["default"]}]
# Readiness is probed in the default profile's own home: live ``beta`` must not vouch for it.
assert {"home": homes["default"]} in homes["_ready_probes"]
# The authorizing generation is consumed and the NEW PID is attested in the same profile home,
# so a death after this CLI exits stays visible to the next update.
reattested = json.loads(marker.read_text(encoding="utf-8"))
assert (reattested["pids"], reattested["generation"] != generation) == ([4242], True)
assert "cold_start_profiles" not in token
assert token["relaunched_profiles"] == ["beta"]
assert token["resume_needed"] is False
out = capsys.readouterr().out
assert "Gateway profile default started via cold-start after update (PID: 4242)" in out
@pytest.mark.windows_only
def test_every_dead_attested_profile_is_cold_started_when_nothing_runs(monkeypatch, tmp_path):
"""Nothing running, active profile exited cleanly (plan → None), ``beta`` dead-attested: beta still
gets a token and a spawn. And when BOTH owe a spawn, the fleet-wide active cold-start runs FIRST —
a beta spawned earlier would satisfy its any-live-gateway guard and the active profile would stay down."""
homes = _running_beta_pause_fixture(monkeypatch, tmp_path)
monkeypatch.setattr(update_cmd_windows, "_discover_windows_gateways", lambda: ({}, [], set(), []))
monkeypatch.setattr(update_cmd_windows, "_windows_cold_start_plan", lambda: None)
gateway_windows._write_start_attestation([556], "direct spawn (PID 556)", home=homes["beta"])
beta_marker = homes["beta"] / "state" / "gateway.start-attestation.json"
beta_generation = json.loads(beta_marker.read_text(encoding="utf-8"))["generation"]
token = update_cmd._pause_windows_gateways_for_update()
assert token["cold_start_profiles"] == {"beta": beta_generation}
assert token["profiles"] == {}
order = []
monkeypatch.setattr(gateway_windows, "_spawn_detached", lambda **k: order.append(k.get("home")) or 4242)
monkeypatch.setattr(
cli_main, "_cold_start_windows_gateway_after_update", lambda token=None: order.append("active") or True)
token["cold_start_if_installed"] = True # both owe a spawn
update_cmd._resume_windows_gateways_after_update(token)
assert order == ["active", homes["beta"]]
assert json.loads(beta_marker.read_text(encoding="utf-8"))["generation"] != beta_generation
assert token["resume_needed"] is False
@pytest.mark.windows_only
def test_service_supervised_running_profile_is_not_cold_started(monkeypatch, tmp_path):
"""A profile whose gateway is alive under an SCM service is skipped by the socket pause, so it is
absent from ``token["profiles"]``; it must still count as RUNNING for the per-profile probe or its
live attestation reads as dead and resume spawns a second, unsupervised gateway beside the
restarted service."""
from types import SimpleNamespace
homes = _running_beta_pause_fixture(monkeypatch, tmp_path)
svc_proc = SimpleNamespace(pid=900, profile="beta", path=homes["beta"])
service = SimpleNamespace(name="HermesGw-beta", profile="beta", service_pid=800, gateway_pid=900,
descendant_identities=(), service_create_time=1.0, gateway_create_time=2.0)
monkeypatch.setattr(update_cmd_windows, "_discover_windows_gateways", lambda: ({900: svc_proc}, [service], {900}, [900]))
monkeypatch.setattr(update_cmd_windows, "_request_socket_pauses", lambda *a: ({}, [], []))
monkeypatch.setattr(update_cmd, "_stop_windows_gateway_service", lambda *a, **k: None)
gateway_windows._write_start_attestation([900], "direct spawn (PID 900)", home=homes["beta"])
token = update_cmd._pause_windows_gateways_for_update()
assert token["services"] == ["HermesGw-beta"]
assert token["profiles"] == {}
assert "cold_start_profiles" not in token