"""Host-wide singleton invariants (multiplex-only): one lock per host, staleness is proved.""" import dataclasses import json import os import signal import socket import subprocess import sys import time from pathlib import Path import pytest from gateway import host_rendezvous as hr _CHILD = """ import json, os, sys, time sys.path.insert(0, {tree!r}) from gateway.status import acquire_gateway_runtime_lock from gateway import host_rendezvous as hr print(json.dumps({{ "per_home": acquire_gateway_runtime_lock(), "host": hr.claim_host_lock(hr.ROLE_GATEWAY)[0] is hr.HostLockOutcome.ACQUIRED, }}), flush=True) time.sleep(60) """ # Publishes a serve record + its 0600 token, arms the exit cleanup, then waits to be killed. _SIGTERM_CHILD = """ import sys, time sys.path.insert(0, {tree!r}) from gateway import host_rendezvous as hr assert hr.claim_host_lock(hr.ROLE_SERVE)[0] is hr.HostLockOutcome.ACQUIRED hr.publish_record(hr.ROLE_SERVE, host="127.0.0.1", port=9119, token="live-session-token") hr.cleanup_on_exit(hr.ROLE_SERVE) print("published", flush=True) time.sleep(120) """ @pytest.fixture def host_dir(tmp_path, monkeypatch): monkeypatch.setenv("HERMES_GATEWAY_LOCK_DIR", str(tmp_path / "locks")) yield tmp_path hr.release_host_lock(hr.ROLE_GATEWAY) hr.release_host_lock(hr.ROLE_SERVE) def _tree() -> str: return str(Path(__file__).resolve().parents[2]) def test_two_homes_take_their_own_per_home_lock_but_only_one_host_lock(host_dir, monkeypatch): """The per-home lock is per HERMES_HOME (N profiles = N locks); the host lock is not. This is the whole point of the host layer: before it, a second profile's gateway took its own ``gateway.lock`` and nothing on the machine noticed. """ tree = _tree() home_a, home_b = host_dir / "home_a", host_dir / "home_b" for home in (home_a, home_b): home.mkdir() env = {**os.environ, "HERMES_HOME": str(home_a), "HERMES_GATEWAY_LOCK_DIR": str(host_dir / "locks")} child = subprocess.Popen( [sys.executable, "-c", _CHILD.format(tree=tree)], env=env, cwd=tree, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True) assert child.stdout is not None try: deadline = time.time() + 60 line = "" while not line.strip() and time.time() < deadline: line = child.stdout.readline() first = json.loads(line) assert first == {"per_home": True, "host": True} monkeypatch.setenv("HERMES_HOME", str(home_b)) from gateway import status assert status.acquire_gateway_runtime_lock() is True, "second home must get its OWN lock" assert hr.claim_host_lock(hr.ROLE_GATEWAY)[0] is hr.HostLockOutcome.HELD_BY_OTHER, \ "host lock is not per-home" finally: child.kill() child.wait(timeout=10) from gateway import status status.release_gateway_runtime_lock() @pytest.mark.parametrize( "pid,create_time", [(2**22 - 1, 1.0), (os.getpid(), 1.0)], ids=["dead-pid", "creation-time-mismatch"], ) def test_stale_record_is_never_attachable(host_dir, pid, create_time): """A dead PID and a live PID from another incarnation are both stale — attaching to either dials whatever now owns that port.""" record = hr.HostRecord( role=hr.ROLE_SERVE, pid=pid, create_time=create_time, host="127.0.0.1", port=9119, protocol_version=hr.HOST_PROTOCOL_VERSION, token_fingerprint="", profiles=("default",), updated_at="2026-01-01T00:00:00+00:00") hr.record_path(hr.ROLE_SERVE).parent.mkdir(parents=True, exist_ok=True) hr.record_path(hr.ROLE_SERVE).write_text(json.dumps(record.to_json()), encoding="utf-8") assert hr.record_is_stale(record) is True assert hr.read_record(hr.ROLE_SERVE) is None assert hr.read_record(hr.ROLE_SERVE, include_stale=True) is not None @pytest.mark.platforms("posix") def test_live_host_record_survives_wall_clock_create_time_drift(host_dir, monkeypatch): """WSL can shift psutil.create_time() for the same live PID while /proc start ticks stay fixed. The host record must keep recognizing that owner, but a changed stable start fingerprint must still be rejected as PID reuse. """ record = hr.publish_record(hr.ROLE_GATEWAY, profiles=("default",), home=str(host_dir)) assert record is not None # Model the exact WSL failure family: the recorded epoch create_time drifted away from the # live process's, while the stable start fingerprint is unchanged. drifted = dataclasses.replace(record, create_time=record.create_time - 5) assert hr.record_is_stale(drifted) is False assert hr.liveness_is_proven(drifted) is True assert record.start_time is not None from gateway import status monkeypatch.setattr( status, "get_process_start_time", lambda _pid: record.start_time + status.START_TIME_DRIFT_TOLERANCE + 1, ) assert hr.record_is_stale(record) is True @pytest.mark.platforms("posix") def test_reused_early_pid_after_reboot_is_not_the_gateway(host_dir): """Boot-relative start ticks reset on reboot and collide within the 2s drift tolerance. A legacy record (no bootEpoch) whose createTime predates this boot, with nobody holding the host lock, cannot describe a live process, and judging it creates no lock file. A record dated inside this boot stays live however far the live create_time reading has since drifted from it (#117505), and a record with no incarnation proof at all never vouches for the pid. """ import psutil from gateway.status import get_process_start_time child = subprocess.Popen(["sleep", "30"]) try: start = get_process_start_time(child.pid) assert start is not None boot = psutil.boot_time() record = hr.HostRecord( role=hr.ROLE_GATEWAY, pid=child.pid, create_time=boot - 3600.0, start_time=int(start) + 10, host="", port=None, protocol_version=hr.HOST_PROTOCOL_VERSION, token_fingerprint="", profiles=("default",), updated_at="2026-01-01T00:00:00+00:00", ) assert hr.liveness_is_proven(record) is False assert hr.record_is_stale(record) is True assert not hr.lock_path(hr.ROLE_GATEWAY).exists() drifted = dataclasses.replace(record, create_time=boot + 1.0) assert hr.liveness_is_proven(drifted) is True assert hr.record_is_stale(drifted) is False unprovable = dataclasses.replace(record, create_time=None, start_time=None) assert hr.liveness_is_proven(unprovable) is False assert hr.record_is_stale(unprovable) is True finally: child.kill() child.wait(timeout=10) @pytest.mark.platforms("posix") def test_live_legacy_owner_survives_a_clock_step_past_its_create_time(host_dir): """A forward wall-clock step moves boot time past a running owner's saved createTime (boot time is wall clock minus uptime). The host lock is held, so the owner stays live.""" import psutil from gateway.status import get_process_start_time env = {**os.environ, "HERMES_HOME": str(host_dir / "home"), "HERMES_GATEWAY_LOCK_DIR": str(host_dir / "locks")} child = subprocess.Popen( [sys.executable, "-c", _CHILD.format(tree=_tree())], env=env, cwd=_tree(), stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True) assert child.stdout is not None try: assert json.loads(child.stdout.readline())["host"] is True start = get_process_start_time(child.pid) assert start is not None record = hr.HostRecord( role=hr.ROLE_GATEWAY, pid=child.pid, create_time=psutil.boot_time() - 3600.0, start_time=int(start), host="", port=None, protocol_version=hr.HOST_PROTOCOL_VERSION, token_fingerprint="", profiles=("default",), updated_at="2026-01-01T00:00:00+00:00", ) assert hr.liveness_is_proven(record) is True assert hr.record_is_stale(record) is False finally: child.kill() child.wait(timeout=10) @pytest.mark.platforms("linux") def test_boot_epoch_mismatch_rejects_a_tick_match(host_dir): record = hr.publish_record(hr.ROLE_GATEWAY, profiles=("default",), home=str(host_dir)) assert record is not None and record.boot_epoch foreign = dataclasses.replace(record, boot_epoch=record.boot_epoch + "-other-boot") assert hr.liveness_is_proven(record) is True assert hr.liveness_is_proven(foreign) is False assert hr.record_is_stale(foreign) is True @pytest.mark.platforms("linux") def test_recorded_epoch_with_unreadable_current_epoch_is_unprovable(host_dir, monkeypatch): record = hr.publish_record(hr.ROLE_GATEWAY, profiles=("default",), home=str(host_dir)) assert record is not None and record.boot_epoch monkeypatch.setattr(hr, "_current_boot_epoch", lambda: "") assert hr._record_incarnation_matches(record) is None assert hr.liveness_is_proven(record) is False assert hr.record_is_stale(record) is False @pytest.mark.platforms("posix") def test_empty_boot_epoch_does_not_stale_a_live_owner(host_dir): record = hr.publish_record(hr.ROLE_GATEWAY, profiles=("default",), home=str(host_dir)) assert record is not None empty = dataclasses.replace(record, boot_epoch="") assert hr.liveness_is_proven(empty) is True assert hr.record_is_stale(empty) is False @pytest.mark.platforms("posix") # POSIX signal disposition def test_sigterm_removes_the_record_and_its_live_session_token(host_dir): """SIGTERM is the NORMAL stop (systemd stop, docker stop, the update relaunch) and it does not run ``atexit``: the record outlived its process and the 0600 token kept a LIVE session token on disk indefinitely, so the next launch attached to something that was gone.""" env = {**os.environ, "HERMES_GATEWAY_LOCK_DIR": str(host_dir / "locks")} child = subprocess.Popen( [sys.executable, "-c", _SIGTERM_CHILD.format(tree=_tree())], env=env, cwd=_tree(), stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True) assert child.stdout is not None try: assert child.stdout.readline().strip() == "published" assert hr.record_path(hr.ROLE_SERVE).exists() assert hr.token_path(hr.ROLE_SERVE).exists() child.send_signal(signal.SIGTERM) child.wait(timeout=30) finally: if child.poll() is None: child.kill() child.wait(timeout=10) assert not hr.record_path(hr.ROLE_SERVE).exists(), "record outlived its process" assert not hr.token_path(hr.ROLE_SERVE).exists(), "live session token outlived its process" assert hr.claim_host_lock(hr.ROLE_SERVE)[0] is hr.HostLockOutcome.ACQUIRED, "host lock not released" def test_unopenable_lock_dir_is_not_reported_as_another_owner(host_dir, monkeypatch): """EROFS/EACCES/ENOSPC is a THIRD outcome: collapsing it into contention told operators another gateway owned the host and sent them hunting a process that never existed.""" blocked = host_dir / "blocked" blocked.write_text("not a directory", encoding="utf-8") monkeypatch.setenv("HERMES_GATEWAY_LOCK_DIR", str(blocked / "locks")) outcome, error = hr.claim_host_lock(hr.ROLE_SERVE) assert outcome is hr.HostLockOutcome.COULD_NOT_OPEN assert isinstance(error, OSError) def test_lock_cache_follows_the_lock_directory(host_dir, monkeypatch, tmp_path): """The handle cache is keyed by (role, resolved path): keyed by role alone it reported "already held" for a directory it had never created, so ``owns_host_lock()`` lied.""" assert hr.claim_host_lock(hr.ROLE_SERVE)[0] is hr.HostLockOutcome.ACQUIRED moved = tmp_path / "other-locks" monkeypatch.setenv("HERMES_GATEWAY_LOCK_DIR", str(moved)) assert hr.owns_host_lock(hr.ROLE_SERVE) is False assert hr.claim_host_lock(hr.ROLE_SERVE)[0] is hr.HostLockOutcome.ACQUIRED assert (moved / "host-serve.lock").exists() hr.release_host_lock(hr.ROLE_SERVE) def test_probe_owner_refuses_a_closed_port_and_a_foreign_listener(host_dir): """The two shapes the record cannot see: the owner's socket is already closed (the graceful-shutdown window) and an unrelated process holds the port.""" import dataclasses closed = socket.socket() closed.bind(("127.0.0.1", 0)) dead_port = closed.getsockname()[1] closed.close() record = hr.HostRecord( role=hr.ROLE_SERVE, pid=os.getpid(), create_time=hr.process_create_time(), host="127.0.0.1", port=dead_port, protocol_version=hr.HOST_PROTOCOL_VERSION, token_fingerprint="", profiles=(), updated_at="2026-01-01T00:00:00+00:00") assert hr.probe_owner(record, timeout=1.0) is None foreign = socket.socket() foreign.bind(("127.0.0.1", 0)) foreign.listen(1) try: assert hr.probe_owner( dataclasses.replace(record, port=foreign.getsockname()[1]), timeout=1.0) is None finally: foreign.close() def test_relative_xdg_state_home_is_ignored(monkeypatch, tmp_path): """XDG spec: a relative ``$XDG_STATE_HOME`` is invalid and must be ignored. Honouring one made the host lock dir CWD-relative, so two serves started from different directories would each take their own "host" lock.""" from gateway import status monkeypatch.delenv("HERMES_GATEWAY_LOCK_DIR", raising=False) monkeypatch.setenv("XDG_STATE_HOME", "relative/state") monkeypatch.setattr(Path, "home", classmethod(lambda cls: tmp_path)) assert status._get_lock_dir().is_absolute() assert status._get_lock_dir() == tmp_path / ".local" / "state" / "hermes" / status._LOCKS_DIRNAME def _gateway_record(*, pid: int, create_time, start_time, home: str) -> hr.HostRecord: return hr.HostRecord( role=hr.ROLE_GATEWAY, pid=pid, create_time=create_time, start_time=start_time, host="", port=None, protocol_version=hr.HOST_PROTOCOL_VERSION, token_fingerprint="", profiles=("default",), updated_at="2026-01-01T00:00:00+00:00", home=home, ) def test_missing_incarnation_fields_do_not_prove_a_live_pid(host_dir): """``_same_incarnation`` treats ``create_time=None`` as a match, so a recycled pid looks live. A record with neither createTime nor startTime therefore cannot prove incarnation — the attach path used that false proof to refuse forever (#138343). Topology already rejected this shape; the shared liveness predicate did not. """ record = _gateway_record( pid=os.getpid(), create_time=None, start_time=None, home=str(host_dir), ) assert hr.liveness_is_proven(record) is False def test_free_host_lock_means_attach_does_not_treat_a_recycled_pid_as_owner(host_dir, monkeypatch): """Every live publisher holds ``host-gateway.lock`` for its lifetime; the OS releases it on death. A leftover record can still name a live pid (recycle, or start-tick collision after reboot). If the flock is free, that pid is not the owner, and ``decide()`` must START instead of the never-expiring transient refuse that wedged supervised units in #138343 / #133744. """ from gateway.host_attach import START, decide, invalidate_host_gateway_cache from gateway.status import get_process_start_time child = subprocess.Popen([sys.executable, "-c", "import time; time.sleep(60)"]) try: # The record must name OUR home: a record from another tenant root starts beside it anyway. home = host_dir / "home" home.mkdir() record = _gateway_record( pid=child.pid, create_time=hr.process_create_time(child.pid), start_time=get_process_start_time(child.pid), home=str(home), ) hr.ensure_host_state_dir() hr.record_path(hr.ROLE_GATEWAY).write_text(json.dumps(record.to_json()), encoding="utf-8") monkeypatch.setattr("gateway.host_attach._identify", lambda home: None) invalidate_host_gateway_cache() decision = decide(home) assert decision.outcome == START assert decision.owner is None finally: child.kill() child.wait(timeout=10) def test_holding_the_host_lock_keeps_a_matching_incarnation_live(host_dir): """The boot-race window: lock is claimed before the socket binds, so a silent socket stays transient.""" assert hr.claim_host_lock(hr.ROLE_GATEWAY)[0] is hr.HostLockOutcome.ACQUIRED record = hr.publish_record(hr.ROLE_GATEWAY, profiles=("default",), home=str(host_dir)) assert record is not None assert hr.host_lock_is_held(hr.ROLE_GATEWAY) is True assert hr.liveness_is_proven(record) is True def test_null_create_time_record_on_a_recycled_pid_starts_instead_of_refusing_forever(host_dir, monkeypatch): """The outage shape: a SIGKILLed owner left ``createTime: null``, its pid now an unrelated process. ``decide()`` must START (take the host over), not the transient REFUSE a supervisor retries forever. """ from gateway.host_attach import START, decide, invalidate_host_gateway_cache child = subprocess.Popen([sys.executable, "-c", "import time; time.sleep(60)"]) try: home = host_dir / "home" home.mkdir() record = _gateway_record(pid=child.pid, create_time=None, start_time=None, home=str(home)) record = dataclasses.replace(record, updated_at="2020-01-01T00:00:00+00:00") hr.ensure_host_state_dir() hr.record_path(hr.ROLE_GATEWAY).write_text(json.dumps(record.to_json()), encoding="utf-8") monkeypatch.setattr("gateway.host_attach._identify", lambda home: None) invalidate_host_gateway_cache() assert decide(home).outcome == START finally: child.kill() child.wait(timeout=10) def test_discard_dead_record_retracts_a_live_looking_pid_only_once_the_host_lock_is_free(host_dir): """`gateway restart --all` retracts a corpse's record; a recycled pid can still match its incarnation. Lock held: the recorded owner may be alive, keep the record. Lock free: the owner is gone. """ from gateway.status import get_process_start_time child = subprocess.Popen([sys.executable, "-c", "import time; time.sleep(60)"]) try: record = _gateway_record( pid=child.pid, create_time=hr.process_create_time(child.pid), start_time=get_process_start_time(child.pid), home=str(host_dir), ) hr.ensure_host_state_dir() hr.record_path(hr.ROLE_GATEWAY).write_text(json.dumps(record.to_json()), encoding="utf-8") assert hr.claim_host_lock(hr.ROLE_GATEWAY)[0] is hr.HostLockOutcome.ACQUIRED assert hr.discard_dead_record(hr.ROLE_GATEWAY) is False assert hr.record_path(hr.ROLE_GATEWAY).exists() hr.release_host_lock(hr.ROLE_GATEWAY) assert hr.discard_dead_record(hr.ROLE_GATEWAY) is True assert not hr.record_path(hr.ROLE_GATEWAY).exists() finally: child.kill() child.wait(timeout=10) # Holds the host lock like a running gateway until killed. _HOLDER_CHILD = """ import sys, time sys.path.insert(0, {tree!r}) from gateway import host_rendezvous as hr assert hr.claim_host_lock(hr.ROLE_GATEWAY)[0] is hr.HostLockOutcome.ACQUIRED print("held", flush=True) time.sleep(60) """ @pytest.mark.platforms("posix") def test_another_process_holding_the_host_lock_is_observed_and_its_exit_frees_the_file(host_dir): """The parent never owns the lock here, so every answer comes from probing the lock file. Held by a child: the probe reports held and restart cleanup keeps a live-looking record. Child killed: the same lock file still exists, reads free, and the cleanup retracts the record. """ from gateway.status import get_process_start_time tree = _tree() sleeper = subprocess.Popen([sys.executable, "-c", "import time; time.sleep(60)"]) holder = subprocess.Popen( [sys.executable, "-u", "-c", _HOLDER_CHILD.format(tree=tree)], cwd=tree, env={**os.environ, "HERMES_GATEWAY_LOCK_DIR": str(host_dir / "locks")}, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True) try: assert holder.stdout.readline().strip() == "held" record = _gateway_record( pid=sleeper.pid, create_time=hr.process_create_time(sleeper.pid), start_time=get_process_start_time(sleeper.pid), home=str(host_dir), ) hr.ensure_host_state_dir() hr.record_path(hr.ROLE_GATEWAY).write_text(json.dumps(record.to_json()), encoding="utf-8") assert not hr.owns_host_lock(hr.ROLE_GATEWAY) assert hr.host_lock_is_held(hr.ROLE_GATEWAY) is True assert hr.discard_dead_record(hr.ROLE_GATEWAY) is False assert hr.record_path(hr.ROLE_GATEWAY).exists() holder.kill() holder.wait(timeout=10) assert hr.lock_path(hr.ROLE_GATEWAY).exists() assert hr.host_lock_is_held(hr.ROLE_GATEWAY) is False assert hr.discard_dead_record(hr.ROLE_GATEWAY) is True assert not hr.record_path(hr.ROLE_GATEWAY).exists() finally: holder.kill() sleeper.kill() holder.wait(timeout=10) sleeper.wait(timeout=10) # Claims the host lock and publishes on demand, like a gateway starting during a restart. _CLAIMER_CHILD = """ import json, sys sys.path.insert(0, {tree!r}) from gateway import host_rendezvous as hr print("ready", flush=True) for line in sys.stdin: won = hr.claim_host_lock(hr.ROLE_GATEWAY)[0] is hr.HostLockOutcome.ACQUIRED if won: hr.publish_record(hr.ROLE_GATEWAY, profiles=("default",), token="new-owner-token") print(json.dumps(won), flush=True) """ @pytest.mark.platforms("posix") def test_restart_cleanup_never_deletes_a_gateway_that_claimed_the_lock_mid_cleanup(host_dir, monkeypatch): """A free-lock probe released before the unlink let a new gateway publish in the gap. The restart's record cleanup then deleted the NEW owner's record and token while it held the host lock. A gateway holding the lock must always have its record and token on disk. """ from gateway.status import get_process_start_time from hermes_cli.gateway_host_record import _discard_dead_host_record tree = _tree() sleeper = subprocess.Popen([sys.executable, "-c", "import time; time.sleep(60)"]) claimer = subprocess.Popen( [sys.executable, "-u", "-c", _CLAIMER_CHILD.format(tree=tree)], cwd=tree, env={**os.environ, "HERMES_GATEWAY_LOCK_DIR": str(host_dir / "locks")}, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True) try: assert claimer.stdout.readline().strip() == "ready" def claim() -> bool: claimer.stdin.write("claim\n") claimer.stdin.flush() return json.loads(claimer.stdout.readline()) record = _gateway_record( pid=sleeper.pid, create_time=hr.process_create_time(sleeper.pid), start_time=get_process_start_time(sleeper.pid), home=str(host_dir), ) hr.ensure_host_state_dir() hr.record_path(hr.ROLE_GATEWAY).write_text(json.dumps(record.to_json()), encoding="utf-8") won_in_gap = [] real_unlink = Path.unlink def unlink_after_a_racing_claim(self, *args, **kwargs): if not won_in_gap: won_in_gap.append(claim()) return real_unlink(self, *args, **kwargs) with monkeypatch.context() as patch: patch.setattr(Path, "unlink", unlink_after_a_racing_claim) _discard_dead_host_record() assert won_in_gap, "cleanup never reached the unlink" if not won_in_gap[0]: assert claim() is True, "the cleanup must release the host lock it took" current = hr.read_record(hr.ROLE_GATEWAY, include_stale=True) assert current is not None and current.pid == claimer.pid assert hr.token_path(hr.ROLE_GATEWAY).exists() finally: claimer.kill() sleeper.kill() claimer.wait(timeout=10) sleeper.wait(timeout=10)