diff --git a/gateway/platforms/base.py b/gateway/platforms/base.py index c963f875758e..7238c2c50952 100644 --- a/gateway/platforms/base.py +++ b/gateway/platforms/base.py @@ -2448,6 +2448,12 @@ class BasePlatformAdapter(ABC): self._fatal_error_message: Optional[str] = None self._fatal_error_retryable = True self._fatal_error_handler: Optional[Callable[["BasePlatformAdapter"], Awaitable[None] | None]] = None + # Cross-HERMES_HOME token takeover is armed by GatewayRunner only for + # an adapter's initial connect during an explicit ``gateway run + # --replace`` startup. Ordinary starts and every reconnect fail safe + # through the existing retryable conflict path. + self._platform_lock_takeover_allowed = False + self._platform_lock_takeover_attempted = False # Track active message handlers per session for interrupt support. # _active_sessions stores the per-session interrupt Event; _session_tasks @@ -2847,8 +2853,18 @@ class BasePlatformAdapter(ABC): await result def _acquire_platform_lock(self, scope: str, identity: str, resource_desc: str) -> bool: - """Acquire a scoped lock for this adapter. Returns True on success.""" - from gateway.status import acquire_scoped_lock + """Acquire a scoped lock for this adapter. Returns True on success. + + A live cross-HERMES_HOME holder may be replaced only when the runner + explicitly arms this adapter for its initial ``--replace`` connect. + The status module validates PID/start-time/home ownership, places the + marker in the target's home, and performs the bounded termination. + """ + from gateway.status import ( + acquire_scoped_lock, + take_over_scoped_lock_holder, + ) + self._platform_lock_scope = scope self._platform_lock_identity = identity acquired, existing = acquire_scoped_lock( @@ -2856,6 +2872,42 @@ class BasePlatformAdapter(ABC): ) if acquired: return True + + takeover_allowed = bool( + getattr(self, "_platform_lock_takeover_allowed", False) + ) + takeover_attempted = bool( + getattr(self, "_platform_lock_takeover_attempted", False) + ) + if takeover_allowed and not takeover_attempted and isinstance(existing, dict): + # Consume the authority before doing any I/O: one adapter connect + # gets at most one termination attempt, even if lock re-acquire or + # later initialization fails. + self._platform_lock_takeover_allowed = False + self._platform_lock_takeover_attempted = True + owner_pid = take_over_scoped_lock_holder(existing) + if owner_pid is not None: + logger.warning( + "[%s] %s was held by gateway PID %d — explicit --replace " + "handoff completed", + self.name, + resource_desc, + owner_pid, + ) + acquired, existing = acquire_scoped_lock( + scope, + identity, + metadata={"platform": self.platform.value}, + ) + if acquired: + logger.info( + "[%s] Acquired %s after taking over PID %d", + self.name, + resource_desc, + owner_pid, + ) + return True + owner_pid = existing.get('pid') if isinstance(existing, dict) else None message = ( f'{resource_desc} already in use' diff --git a/gateway/run.py b/gateway/run.py index 01fddf2dfaaa..b27d231f9e2b 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -3097,6 +3097,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew _loop_heartbeat_task: Optional["asyncio.Task"] = None _gateway_started_at: float = 0.0 _shutdown_watchdog_done: Optional["threading.Event"] = None + _platform_lock_takeover_on_start: bool = False def __init__(self, config: Optional[GatewayConfig] = None): global _gateway_runner_ref @@ -3256,6 +3257,10 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew # with the synthetic resume turns for the same session. The queued # events drain only after all startup resume tasks have finished. self._startup_restore_in_progress = False + # Set by start_gateway() only for an explicit ``--replace`` launch. + # _connect_initial_adapter_with_timeout scopes it to each adapter's + # cold-start connect and removes it before any reconnect can run. + self._platform_lock_takeover_on_start = False self._startup_restore_queue: List[MessageEvent] = [] self._startup_restore_tasks: List[asyncio.Task] = [] # LRU cache of live SessionSources keyed by session_key. Used by @@ -3841,6 +3846,22 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew f"{platform.value} connect timed out after {timeout:g}s" ) from exc + async def _connect_initial_adapter_with_timeout(self, adapter, platform) -> bool: + """Connect one cold-start adapter with tightly scoped replace intent. + + The capability is visible only while this initial connect is awaited. + Reconnects call ``_connect_adapter_with_timeout`` directly and adapters + also default to deny, so a later network recovery can never evict a + healthy token holder. + """ + adapter._platform_lock_takeover_allowed = bool( + self._platform_lock_takeover_on_start + ) + try: + return await self._connect_adapter_with_timeout(adapter, platform) + finally: + adapter._platform_lock_takeover_allowed = False + @property def should_exit_cleanly(self) -> bool: return self._exit_cleanly @@ -7697,7 +7718,9 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew error_message=None, ) try: - success = await self._connect_adapter_with_timeout(adapter, platform) + success = await self._connect_initial_adapter_with_timeout( + adapter, platform + ) if await self._abort_startup_if_shutdown_requested(adapter, platform): return True if success: @@ -7805,6 +7828,10 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew return True except Exception as e: logger.error("Secondary-profile adapter startup failed: %s", e, exc_info=True) + finally: + # Startup authority is one phase, not a persistent runner mode. + # From this point onward every adapter retry is non-evicting. + self._platform_lock_takeover_on_start = False # A platform we skipped on the primary for a missing credential was # supposed to be picked up by a secondary profile that owns the token. @@ -9503,7 +9530,9 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew try: with _profile_runtime_scope(profile_home): - success = await self._connect_adapter_with_timeout(adapter, platform) + success = await self._connect_initial_adapter_with_timeout( + adapter, platform + ) if success: profile_map[platform] = adapter connected += 1 @@ -22637,6 +22666,10 @@ async def start_gateway(config: Optional[GatewayConfig] = None, replace: bool = logging.getLogger().setLevel(_stderr_level) runner = GatewayRunner(config) + # ``--replace`` is explicit startup authority, not a durable reconnect + # policy. GatewayRunner scopes this bit to cold adapter connects and clears + # it before the background reconnect watcher starts. + runner._platform_lock_takeover_on_start = bool(replace) # Track whether an unexpected signal initiated the shutdown. When an # unexpected SIGTERM kills the gateway, we exit non-zero so service diff --git a/gateway/status.py b/gateway/status.py index e265d380df91..544929ed53fc 100644 --- a/gateway/status.py +++ b/gateway/status.py @@ -142,6 +142,18 @@ def _get_process_hermes_home() -> Path: return _get_platform_default_hermes_home() +def _canonical_hermes_home(path: Path | str) -> Path: + """Return a stable absolute HERMES_HOME path for persisted identity data.""" + return Path(path).expanduser().resolve(strict=False) + + +def _same_hermes_home(left: Path | str, right: Path | str) -> bool: + """Compare HERMES_HOME paths with the host platform's case semantics.""" + return os.path.normcase(str(_canonical_hermes_home(left))) == os.path.normcase( + str(_canonical_hermes_home(right)) + ) + + def _get_pid_path() -> Path: """Return the path to the gateway PID file, respecting HERMES_HOME.""" home = _get_process_hermes_home() @@ -499,6 +511,11 @@ def _build_pid_record() -> dict: "kind": _GATEWAY_KIND, "argv": list(sys.argv), "start_time": _get_process_start_time(os.getpid()), + # Scoped credential locks are machine-global rather than + # HERMES_HOME-local. Persist the owning gateway's process home so an + # explicit cross-profile --replace can place its planned-takeover + # marker where the target process will actually read it. + "hermes_home": str(_canonical_hermes_home(_get_process_hermes_home())), } @@ -1356,10 +1373,14 @@ _PLANNED_STOP_MARKER_FILENAME = ".gateway-planned-stop.json" _PLANNED_STOP_MARKER_TTL_S = 60 -def _get_takeover_marker_path() -> Path: - """Return the path to the --replace takeover marker file.""" - home = _get_process_hermes_home() - return home / _TAKEOVER_MARKER_FILENAME +def _get_takeover_marker_path(hermes_home: Optional[Path] = None) -> Path: + """Return the path to the --replace takeover marker file. + + ``hermes_home`` is supplied only for a verified cross-home handoff. The + target process always consumes the marker from its own process-level home. + """ + home = hermes_home or _get_process_hermes_home() + return _canonical_hermes_home(home) / _TAKEOVER_MARKER_FILENAME def _get_planned_stop_marker_path() -> Path: @@ -1406,21 +1427,24 @@ def _consume_pid_marker_for_self( pass return False - # Cross-profile guard (#29092): reject markers written by a gateway - # running under a different HERMES_HOME. When two profile gateway - # services share the same default ~/.hermes (HERMES_HOME not set - # distinctly), the marker path resolves to the same file for both. A - # --replace from profile B could land in profile A's marker, match on - # PID + start_time by coincidence of a shared PID namespace, and make - # profile A exit 0 — only to be revived by systemd Restart=always, - # which then races the replacer again, flapping indefinitely. The - # field is absent in markers written by older Hermes versions; treat - # absent as "same home" so old markers and single-profile setups are - # unaffected. Leave a mismatched marker in place so the correct - # profile can still consume it. - replacer_home = record.get("replacer_hermes_home") - if replacer_home is not None and replacer_home != str(_get_process_hermes_home()): - return False + # Cross-profile guard (#29092): new markers explicitly name the verified + # TARGET home. That permits a deliberate cross-HERMES_HOME --replace while + # ensuring a marker accidentally written into another profile's directory + # is ignored. Legacy markers have no target field, so retain the original + # same-replacer-home rule for backwards compatibility. + our_home = _get_process_hermes_home() + target_home = record.get("target_hermes_home") + if target_home is not None: + if not isinstance(target_home, str) or not _same_hermes_home( + target_home, our_home + ): + return False + else: + replacer_home = record.get("replacer_hermes_home") + if replacer_home is not None and not _same_hermes_home( + replacer_home, our_home + ): + return False our_pid = os.getpid() our_start_time = _get_process_start_time(our_pid) @@ -1451,27 +1475,45 @@ def _consume_pid_marker_for_self( return matches -def write_takeover_marker(target_pid: int) -> bool: +def write_takeover_marker( + target_pid: int, + *, + target_home: Optional[Path] = None, + target_start_time: Any = _UNSET, +) -> bool: """Record that ``target_pid`` is being replaced by the current process. Captures the target's ``start_time`` so that PID reuse after the target exits cannot later match the marker. Also records the replacer's PID and a UTC timestamp for TTL-based staleness checks. - Returns True on successful write, False on any failure. The caller - should proceed with the SIGTERM even if the write fails (the marker - is a best-effort signal, not a correctness requirement). + A verified scoped-lock handoff supplies ``target_home`` and the already + validated ``target_start_time`` so the marker is written into the target + gateway's HERMES_HOME rather than the replacer's. Same-home callers omit + both arguments and preserve the historical behavior. + + Returns True on successful write, False on any failure. Historical + same-home callers may treat the marker as best effort. Cross-home callers + must fail closed because the target's supervisor could otherwise revive it + without recognizing the handoff. """ try: - target_start_time = _get_process_start_time(target_pid) + marker_home = _canonical_hermes_home( + target_home or _get_process_hermes_home() + ) + if target_start_time is _UNSET: + target_start_time = _get_process_start_time(target_pid) record = { "target_pid": target_pid, "target_start_time": target_start_time, + "target_hermes_home": str(marker_home), "replacer_pid": os.getpid(), - "replacer_hermes_home": str(_get_process_hermes_home()), + "replacer_hermes_home": str( + _canonical_hermes_home(_get_process_hermes_home()) + ), "written_at": _utc_now_iso(), } - _write_json_file(_get_takeover_marker_path(), record) + _write_json_file(_get_takeover_marker_path(marker_home), record) return True except (OSError, PermissionError): return False @@ -1496,14 +1538,182 @@ def consume_takeover_marker_for_self() -> bool: ) -def clear_takeover_marker() -> None: +def clear_takeover_marker(target_home: Optional[Path] = None) -> None: """Remove the takeover marker unconditionally. Safe to call repeatedly.""" try: - _get_takeover_marker_path().unlink(missing_ok=True) + _get_takeover_marker_path(target_home).unlink(missing_ok=True) except OSError: pass +def _validated_scoped_lock_gateway_owner( + record: dict[str, Any], +) -> Optional[tuple[int, int, Path]]: + """Resolve a live scoped-lock owner to a verified gateway identity. + + A machine-global scoped-lock file is only a claim; it is not sufficient + authority to terminate a process or choose a marker destination. Require + the lock record, the target HERMES_HOME's gateway PID record, and the live + OS process to agree on PID, start-time fingerprint, gateway identity, and + process home. Missing legacy metadata fails closed and leaves the normal + retryable lock-conflict path in charge. + """ + if not isinstance(record, dict) or not _record_looks_like_gateway(record): + return None + + try: + owner_pid = int(record["pid"]) + except (KeyError, TypeError, ValueError): + return None + if owner_pid <= 0 or owner_pid == os.getpid(): + return None + + owner_start_time = record.get("start_time") + if not isinstance(owner_start_time, int) or isinstance(owner_start_time, bool): + return None + + raw_home = record.get("hermes_home") + if not isinstance(raw_home, str) or not raw_home.strip(): + return None + if not Path(raw_home).expanduser().is_absolute(): + return None + target_home = _canonical_hermes_home(raw_home) + + if not _pid_exists(owner_pid): + return None + live_start_time = _get_process_start_time(owner_pid) + if live_start_time is None or live_start_time != owner_start_time: + return None + + live_cmdline = _read_process_cmdline(owner_pid) + if live_cmdline is not None and not looks_like_gateway_runtime_command_line( + live_cmdline + ): + return None + + pid_record = _read_json_file(target_home / "gateway.pid") + if not isinstance(pid_record, dict) or not _record_looks_like_gateway(pid_record): + return None + try: + pid_record_pid = int(pid_record["pid"]) + except (KeyError, TypeError, ValueError): + return None + if pid_record_pid != owner_pid or pid_record.get("start_time") != owner_start_time: + return None + + pid_record_home = pid_record.get("hermes_home") + if not isinstance(pid_record_home, str) or not _same_hermes_home( + pid_record_home, target_home + ): + return None + + return owner_pid, owner_start_time, target_home + + +def _scoped_lock_owner_state(owner_pid: int, owner_start_time: int) -> str: + """Return ``same``, ``exited``, or ``unknown`` for a validated owner.""" + if not _pid_exists(owner_pid): + return "exited" + live_start_time = _get_process_start_time(owner_pid) + if live_start_time is None: + return "unknown" + if live_start_time != owner_start_time: + # The original owner exited and the OS recycled its PID. Never signal + # the replacement process. + return "exited" + return "same" + + +def _wait_for_scoped_lock_owner_exit( + owner_pid: int, + owner_start_time: int, + *, + attempts: int, + delay: float, +) -> tuple[bool, bool]: + """Return ``(exited, safe_to_force)`` after bounded identity-aware waits.""" + for _ in range(max(0, attempts)): + state = _scoped_lock_owner_state(owner_pid, owner_start_time) + if state == "exited": + return True, False + if state == "unknown": + return False, False + time.sleep(max(0.0, delay)) + return False, _scoped_lock_owner_state(owner_pid, owner_start_time) == "same" + + +def take_over_scoped_lock_holder( + record: dict[str, Any], + *, + graceful_attempts: int = 20, + force_attempts: int = 20, +) -> Optional[int]: + """Terminate one verified scoped-lock holder for explicit ``--replace``. + + Returns the original owner PID only after that exact PID/start-time identity + has exited (or ``terminate_pid`` reports it already gone). Validation or + marker-write failure returns ``None`` without signalling anything. This is + deliberately stricter than the same-home PID-file replacement path: a + cross-home handoff must place a consumable marker in the target's home or a + service supervisor could revive the target and start a flap loop. + """ + owner = _validated_scoped_lock_gateway_owner(record) + if owner is None: + return None + owner_pid, owner_start_time, target_home = owner + + if not write_takeover_marker( + owner_pid, + target_home=target_home, + target_start_time=owner_start_time, + ): + return None + + try: + state = _scoped_lock_owner_state(owner_pid, owner_start_time) + if state == "exited": + return owner_pid + if state != "same": + return None + + try: + terminate_pid(owner_pid, force=False) + except ProcessLookupError: + return owner_pid + except (PermissionError, OSError): + return None + + exited, safe_to_force = _wait_for_scoped_lock_owner_exit( + owner_pid, + owner_start_time, + attempts=graceful_attempts, + delay=0.5, + ) + if exited: + return owner_pid + if not safe_to_force: + return None + + try: + terminate_pid(owner_pid, force=True) + except ProcessLookupError: + return owner_pid + except (PermissionError, OSError): + return None + + exited, _ = _wait_for_scoped_lock_owner_exit( + owner_pid, + owner_start_time, + attempts=force_attempts, + delay=0.25, + ) + return owner_pid if exited else None + finally: + # The target normally consumes its marker from the signal handler. + # Clean up any remainder after an already-gone/forced/failed handoff. + clear_takeover_marker(target_home) + + def write_planned_stop_marker(target_pid: int) -> bool: """Record that ``target_pid`` is being stopped intentionally. diff --git a/tests/gateway/test_runner_startup_failures.py b/tests/gateway/test_runner_startup_failures.py index 4a1343ae7683..e8bb72b12e4d 100644 --- a/tests/gateway/test_runner_startup_failures.py +++ b/tests/gateway/test_runner_startup_failures.py @@ -156,6 +156,7 @@ async def test_start_gateway_verbosity_imports_redacting_formatter(monkeypatch, self.adapters = {} async def start(self): + assert self._platform_lock_takeover_on_start is False return True async def stop(self): @@ -191,6 +192,7 @@ async def test_start_gateway_replace_force_uses_terminate_pid(monkeypatch, tmp_p self.adapters = {} async def start(self): + assert self._platform_lock_takeover_on_start is True return True async def stop(self): diff --git a/tests/gateway/test_stale_platform_lock_retryable.py b/tests/gateway/test_stale_platform_lock_retryable.py index 39f1ab2c3882..e62851852321 100644 --- a/tests/gateway/test_stale_platform_lock_retryable.py +++ b/tests/gateway/test_stale_platform_lock_retryable.py @@ -1,5 +1,6 @@ -"""Regression test for #54167 — stale platform lock must be retryable. +"""Regression tests for platform-lock acquire behavior. +#54167 — stale platform lock must be retryable. When a gateway process is killed (SIGKILL, crash) during Telegram initialization, the scoped lock file survives. On next startup, ``acquire_scoped_lock()`` detects the stale lock and deletes it, but may @@ -11,10 +12,11 @@ process grab the lock first). so the reconnect watcher can retry after a delay — not permanently kill the platform. -Contract asserted here ----------------------- -``_set_fatal_error`` is called with ``retryable=True`` when lock -acquisition fails, regardless of the reason. +#65176 — a live gateway token conflict may attempt one-shot takeover only +during the initial connect of an explicit ``gateway run --replace`` startup. +``gateway run --replace`` only kills same-HERMES_HOME PID-file holders. +A normal start or reconnect must retain the retryable conflict behavior and +must never evict the active holder. """ from typing import Any, Dict @@ -23,6 +25,7 @@ from unittest.mock import MagicMock, patch import pytest from gateway.platforms.base import BasePlatformAdapter +from gateway.run import GatewayRunner class _StubAdapter(BasePlatformAdapter): @@ -54,6 +57,8 @@ def adapter(): obj._fatal_error_handler = None obj._platform_lock_scope = None obj._platform_lock_identity = None + obj._platform_lock_takeover_allowed = False + obj._platform_lock_takeover_attempted = False obj._status_write_logged = None return obj @@ -71,3 +76,130 @@ def test_stale_lock_failure_is_retryable(adapter): assert result is False assert adapter._fatal_error_retryable is True assert adapter._fatal_error_code == "telegram-bot-token_lock" + + +def test_explicit_replace_takeover_reacquires_lock_once(adapter): + """Initial explicit --replace may hand off and re-acquire once (#65176).""" + existing = { + "pid": 4242, + "kind": "hermes-gateway", + "argv": ["hermes", "gateway", "run"], + "start_time": 123, + } + acquire = MagicMock(side_effect=[(False, existing), (True, None)]) + adapter._platform_lock_takeover_allowed = True + + with patch("gateway.status.acquire_scoped_lock", acquire), patch( + "gateway.status.take_over_scoped_lock_holder", + return_value=4242, + ) as takeover, patch.object( + adapter, "_write_runtime_status_safe" + ): + result = adapter._acquire_platform_lock( + "telegram-bot-token", "test-token", "Telegram bot token" + ) + + assert result is True + assert adapter._platform_lock_takeover_allowed is False + assert adapter._platform_lock_takeover_attempted is True + takeover.assert_called_once_with(existing) + assert acquire.call_count == 2 + + +def test_normal_connect_conflict_never_attempts_takeover(adapter): + """A normal start/reconnect cannot evict the current token holder.""" + existing = { + "pid": 5555, + "kind": "hermes-gateway", + "argv": ["hermes", "gateway", "run"], + "start_time": 123, + } + with patch( + "gateway.status.acquire_scoped_lock", + return_value=(False, existing), + ), patch( + "gateway.status.take_over_scoped_lock_holder", + ) as takeover, patch.object( + adapter, "_write_runtime_status_safe" + ): + result = adapter._acquire_platform_lock( + "telegram-bot-token", "test-token", "Telegram bot token" + ) + + assert result is False + takeover.assert_not_called() + assert adapter._platform_lock_takeover_attempted is False + assert adapter._fatal_error_retryable is True + + +def test_failed_explicit_takeover_consumes_authority(adapter): + """A failed handoff is not retried by a later acquire on the same adapter.""" + existing = { + "pid": 7777, + "kind": "hermes-gateway", + "argv": ["hermes", "gateway", "run"], + "start_time": 456, + } + adapter._platform_lock_takeover_allowed = True + + with patch( + "gateway.status.acquire_scoped_lock", + return_value=(False, existing), + ), patch( + "gateway.status.take_over_scoped_lock_holder", + return_value=None, + ) as takeover, patch.object( + adapter, "_write_runtime_status_safe" + ): + first = adapter._acquire_platform_lock( + "telegram-bot-token", "test-token", "Telegram bot token" + ) + second = adapter._acquire_platform_lock( + "telegram-bot-token", "test-token", "Telegram bot token" + ) + + assert first is False + assert second is False + assert adapter._platform_lock_takeover_allowed is False + assert adapter._platform_lock_takeover_attempted is True + takeover.assert_called_once_with(existing) + + +@pytest.mark.asyncio +async def test_runner_scopes_replace_intent_to_initial_connect(): + runner = GatewayRunner.__new__(GatewayRunner) + runner._platform_lock_takeover_on_start = True + adapter = MagicMock() + adapter._platform_lock_takeover_allowed = False + seen = [] + + async def connect(current_adapter, _platform): + seen.append(current_adapter._platform_lock_takeover_allowed) + return True + + runner._connect_adapter_with_timeout = connect + + assert await runner._connect_initial_adapter_with_timeout( + adapter, MagicMock(value="telegram") + ) is True + assert seen == [True] + assert adapter._platform_lock_takeover_allowed is False + + +@pytest.mark.asyncio +async def test_runner_clears_replace_intent_when_initial_connect_raises(): + runner = GatewayRunner.__new__(GatewayRunner) + runner._platform_lock_takeover_on_start = True + adapter = MagicMock() + adapter._platform_lock_takeover_allowed = False + + async def connect(_adapter, _platform): + raise RuntimeError("connect failed") + + runner._connect_adapter_with_timeout = connect + + with pytest.raises(RuntimeError, match="connect failed"): + await runner._connect_initial_adapter_with_timeout( + adapter, MagicMock(value="telegram") + ) + assert adapter._platform_lock_takeover_allowed is False diff --git a/tests/gateway/test_status.py b/tests/gateway/test_status.py index 35b35f32a919..f5d01c791503 100644 --- a/tests/gateway/test_status.py +++ b/tests/gateway/test_status.py @@ -1421,6 +1421,155 @@ class TestTakeoverMarker: assert not marker_path.exists() +class TestScopedLockTakeover: + """Cross-home takeover requires explicit, corroborated process identity.""" + + @staticmethod + def _owner_record(target_home: Path, *, pid: int = 4242, start_time: int = 123): + target_home.mkdir(parents=True, exist_ok=True) + record = { + "pid": pid, + "kind": "hermes-gateway", + "argv": ["python", "-m", "hermes_cli.main", "gateway", "run"], + "start_time": start_time, + "hermes_home": str(target_home), + } + (target_home / "gateway.pid").write_text(json.dumps(record)) + return record + + def test_verified_distinct_home_handoff_marks_target_before_sigterm( + self, tmp_path, monkeypatch + ): + replacer_home = tmp_path / "replacer" + target_home = tmp_path / "target" + replacer_home.mkdir() + monkeypatch.setenv("HERMES_HOME", str(replacer_home)) + record = self._owner_record(target_home) + + alive = iter([True, True, False]) + monkeypatch.setattr(status, "_pid_exists", lambda _pid: next(alive)) + monkeypatch.setattr(status, "_get_process_start_time", lambda _pid: 123) + monkeypatch.setattr( + status, + "_read_process_cmdline", + lambda _pid: "python -m hermes_cli.main gateway run", + ) + calls = [] + + def terminate(pid, *, force=False): + marker_path = target_home / ".gateway-takeover.json" + assert marker_path.exists() + payload = json.loads(marker_path.read_text()) + assert payload["target_hermes_home"] == str(target_home) + assert payload["replacer_hermes_home"] == str(replacer_home) + calls.append((pid, force)) + + monkeypatch.setattr(status, "terminate_pid", terminate) + + owner_pid = status.take_over_scoped_lock_holder( + record, graceful_attempts=1 + ) + + assert owner_pid == 4242 + assert calls == [(4242, False)] + assert not (target_home / ".gateway-takeover.json").exists() + assert not (replacer_home / ".gateway-takeover.json").exists() + + def test_handoff_rejects_uncorroborated_target_home(self, tmp_path, monkeypatch): + target_home = tmp_path / "target" + record = self._owner_record(target_home) + # The lock claims target_home, but that home's PID record names a + # different process identity. + bad_pid_record = dict(record, pid=9999) + (target_home / "gateway.pid").write_text(json.dumps(bad_pid_record)) + + monkeypatch.setattr(status, "_pid_exists", lambda _pid: True) + monkeypatch.setattr(status, "_get_process_start_time", lambda _pid: 123) + monkeypatch.setattr( + status, + "_read_process_cmdline", + lambda _pid: "python -m hermes_cli.main gateway run", + ) + calls = [] + monkeypatch.setattr( + status, "terminate_pid", lambda *args, **kwargs: calls.append(args) + ) + + assert status.take_over_scoped_lock_holder(record) is None + assert calls == [] + assert not (target_home / ".gateway-takeover.json").exists() + + def test_handoff_requires_marker_write_before_termination( + self, tmp_path, monkeypatch + ): + target_home = tmp_path / "target" + record = self._owner_record(target_home) + monkeypatch.setattr(status, "_pid_exists", lambda _pid: True) + monkeypatch.setattr(status, "_get_process_start_time", lambda _pid: 123) + monkeypatch.setattr( + status, + "_read_process_cmdline", + lambda _pid: "python -m hermes_cli.main gateway run", + ) + monkeypatch.setattr(status, "write_takeover_marker", lambda *a, **k: False) + calls = [] + monkeypatch.setattr( + status, "terminate_pid", lambda *args, **kwargs: calls.append(args) + ) + + assert status.take_over_scoped_lock_holder(record) is None + assert calls == [] + + def test_pid_reuse_after_sigterm_is_never_force_killed( + self, tmp_path, monkeypatch + ): + target_home = tmp_path / "target" + record = self._owner_record(target_home) + monkeypatch.setattr(status, "_pid_exists", lambda _pid: True) + starts = iter([123, 123, 999]) + monkeypatch.setattr( + status, "_get_process_start_time", lambda _pid: next(starts) + ) + monkeypatch.setattr( + status, + "_read_process_cmdline", + lambda _pid: "python -m hermes_cli.main gateway run", + ) + calls = [] + monkeypatch.setattr( + status, + "terminate_pid", + lambda pid, *, force=False: calls.append((pid, force)), + ) + + assert status.take_over_scoped_lock_holder( + record, graceful_attempts=1 + ) == 4242 + assert calls == [(4242, False)] + + def test_target_accepts_verified_cross_home_marker(self, tmp_path, monkeypatch): + replacer_home = tmp_path / "replacer" + target_home = tmp_path / "target" + replacer_home.mkdir() + target_home.mkdir() + monkeypatch.setenv("HERMES_HOME", str(replacer_home)) + monkeypatch.setattr(status, "_get_process_start_time", lambda _pid: 100) + + assert status.write_takeover_marker( + os.getpid(), + target_home=target_home, + target_start_time=100, + ) is True + assert not (replacer_home / ".gateway-takeover.json").exists() + assert (target_home / ".gateway-takeover.json").exists() + + # The target process reads its own home. A differing replacer home is + # valid only because the marker explicitly names this target home. + monkeypatch.setenv("HERMES_HOME", str(target_home)) + assert status.consume_takeover_marker_for_self() is True + assert not (target_home / ".gateway-takeover.json").exists() + + class TestPlannedStopMarker: """Tests for intentional service/manual gateway stop markers."""