"""Per-profile lifecycle never stops the shared host or hides an installed profile."""
from pathlib import Path
from types import SimpleNamespace

import pytest


@pytest.fixture
def homes(tmp_path, monkeypatch):
    from hermes_cli import profiles
    root = tmp_path / '.hermes'
    secondary = root / 'profiles' / 'worker'
    secondary.mkdir(parents=True)
    (secondary / 'config.yaml').write_text('model: {default: worker-model}\n')
    (root / 'config.yaml').write_text('gateway: {multiplex_profiles: true}\n')
    monkeypatch.setattr(Path, 'home', lambda: tmp_path)
    monkeypatch.setenv('HERMES_HOME', str(root))
    monkeypatch.setattr(profiles, 'get_active_profile_name', lambda: 'default')
    return root, secondary


@pytest.mark.parametrize('include_parked', [False, True])
def test_parked_is_not_served_but_remains_installed(homes, include_parked):
    from hermes_cli import profiles
    root, secondary = homes
    (secondary / 'gateway.parked').write_text('provisioned offline\n')
    if include_parked:
        actual = profiles.profiles_to_serve(True, include_parked=True)
    else:
        actual = profiles.profiles_to_serve(True)
    assert dict(actual) == ({'default': root, 'worker': secondary} if include_parked else {'default': root})
    assert profiles.parked_marker_path(secondary) == secondary / 'gateway.parked'
    assert profiles.profile_is_parked(secondary)


@pytest.mark.parametrize('multiplex', [False, True])
def test_default_marker_is_ignored_with_one_warning(homes, caplog, multiplex):
    from hermes_cli.profiles import profiles_to_serve
    root, _ = homes
    (root / 'gateway.parked').touch()
    for _ in range(2):
        assert 'default' in dict(profiles_to_serve(multiplex))
    warnings = [r for r in caplog.records if 'gateway.parked' in r.message]
    assert len(warnings) == 1
    assert warnings[0].levelname == 'WARNING'


@pytest.mark.parametrize('verb', ['stop', 'start', 'restart'])
def test_cli_lifecycle_orders_marker_before_socket(homes, monkeypatch, capsys, verb):
    from hermes_cli import gateway as gw
    from gateway import control_socket
    root, secondary = homes
    marker = secondary / 'gateway.parked'
    owner = SimpleNamespace(home=root, profile_label='default', profiles=('default', 'worker'),
                            describe=lambda: 'test host')
    monkeypatch.setenv('HERMES_HOME', str(secondary))
    monkeypatch.setattr(gw, '_current_profile_name', lambda: 'worker')
    monkeypatch.setattr(gw, '_refuse_from_inside_gateway', lambda *a: None)
    monkeypatch.setattr(gw, 'find_gateway_pids', lambda **kw: [])
    monkeypatch.setattr(gw, '_served_by_another_host_gateway', lambda *a: owner)
    monkeypatch.setattr(gw, 'named_profile_served_by_running_multiplexer', lambda *a: True)
    monkeypatch.setattr(gw, '_host_multiplexer_for_all_verb', lambda: owner)
    calls = []

    def unserve(home, name):
        assert home == root and name == 'worker'
        assert marker.exists() is (verb == 'stop')
        calls.append('unserve')
        return {'unserved': name, 'served_profiles': ['default']}

    def serve(home, name):
        assert home == root and name == 'worker'
        assert not marker.exists()
        calls.append('serve')
        return {'served': name, 'served_profiles': ['default', name]}

    monkeypatch.setattr(control_socket, 'request_unserve_profile', unserve, raising=False)
    monkeypatch.setattr(control_socket, 'request_serve_profile_hot', serve, raising=False)
    if verb == 'start':
        marker.touch()
    # Precedence: a standalone profile is never parked, even while a stale host record lists it.
    from hermes_cli.gateway_profile_lifecycle import profile_lifecycle
    (secondary / 'config.yaml').write_text('gateway: {standalone: true}\n')
    assert profile_lifecycle(verb, SimpleNamespace()) is False and calls == []
    assert marker.exists() is (verb == 'start')
    (secondary / 'config.yaml').write_text('model: {default: worker-model}\n')
    getattr(gw, '_cmd_' + verb)(SimpleNamespace())
    assert calls == {'stop': ['unserve'], 'start': ['serve'], 'restart': ['unserve', 'serve']}[verb]
    assert marker.exists() is (verb == 'stop')
    output = capsys.readouterr().out
    assert {'stop': "Profile 'worker' parked; its bots and cron are stopped.",
            'start': "Profile 'worker' served", 'restart': "Profile 'worker' restarted"}[verb] in output


def test_restart_after_stop_unparks_and_serves_the_profile(homes, monkeypatch, capsys):
    from hermes_cli import gateway as gw
    from gateway import control_socket
    root, secondary = homes
    marker = secondary / 'gateway.parked'
    marker.touch()
    owner = SimpleNamespace(home=root, profile_label='default', profiles=('default',),
                            describe=lambda: 'test host')
    monkeypatch.setenv('HERMES_HOME', str(secondary))
    monkeypatch.setattr(gw, '_current_profile_name', lambda: 'worker')
    monkeypatch.setattr(gw, '_refuse_from_inside_gateway', lambda *a: None)
    monkeypatch.setattr(gw, 'find_gateway_pids', lambda **kw: [])
    monkeypatch.setattr(gw, '_served_by_another_host_gateway', lambda *a: None)
    monkeypatch.setattr(gw, 'named_profile_served_by_running_multiplexer', lambda *a: False)
    monkeypatch.setattr(gw, '_host_multiplexer_for_all_verb', lambda: owner)
    calls = []

    def serve(home, name):
        assert home == root and name == 'worker' and not marker.exists()
        calls.append('serve')
        return {'served': name, 'served_profiles': ['default', name]}

    def standalone(*args, **kwargs):
        raise AssertionError('restart fell through to the standalone gateway path')

    monkeypatch.setattr(control_socket, 'request_serve_profile_hot', serve)
    monkeypatch.setattr(gw, '_guard_named_profile_under_multiplexer', standalone)
    gw._cmd_restart(SimpleNamespace())
    assert calls == ['serve']
    assert not marker.exists()
    assert "Profile 'worker' served by the host gateway." in capsys.readouterr().out


def test_force_restart_keeps_parked_profile_gateway_ownership(homes, monkeypatch):
    from hermes_cli import gateway as gw
    _, secondary = homes
    (secondary / 'gateway.parked').touch()
    monkeypatch.setenv('HERMES_HOME', str(secondary))
    monkeypatch.setattr(gw, '_current_profile_name', lambda: 'worker')

    def host_restart(*args, **kwargs):
        raise AssertionError('forced gateway restart was routed through the host')

    monkeypatch.setattr(gw, '_served_by_another_host_gateway', lambda *args: None)
    monkeypatch.setattr(gw, 'named_profile_served_by_running_multiplexer', lambda: False)
    monkeypatch.setattr(gw, '_host_multiplexer_for_all_verb', host_restart)
    monkeypatch.setattr(gw, 'find_gateway_pids', lambda **kwargs: [4242])
    from hermes_cli.gateway_profile_lifecycle import profile_lifecycle
    assert profile_lifecycle('restart', SimpleNamespace()) is False


def test_stop_parks_a_served_profile_instead_of_stopping_the_host(homes, monkeypatch, capsys):
    """The host publishes its own pid into a served profile's runtime record (#141133)."""
    from hermes_cli import gateway as gw
    from gateway import control_socket
    root, secondary = homes
    marker = secondary / 'gateway.parked'
    host_pid = 4242
    owner = SimpleNamespace(home=root, pid=host_pid, profile_label='default',
                            profiles=('default', 'worker'), describe=lambda: 'test host')
    from gateway import host_rendezvous as hr, status
    monkeypatch.setenv('HERMES_HOME', str(secondary))
    monkeypatch.setattr(gw, '_current_profile_name', lambda: 'worker')
    monkeypatch.setattr(gw, '_refuse_from_inside_gateway', lambda *a: None)
    # The real find_gateway_pids: the profile's record names the live host, launched from root.
    monkeypatch.setattr(status, 'get_running_pid', lambda **kw: host_pid)
    monkeypatch.setattr(hr, 'read_record', lambda role, **kw: SimpleNamespace(pid=host_pid, home=str(root)))
    monkeypatch.setattr(hr, 'record_names_live_owner', lambda rec: True)
    monkeypatch.setattr(gw, '_get_service_pids', lambda **kw: [])
    monkeypatch.setattr(gw, '_scan_gateway_pids', lambda *a, **kw: [])
    monkeypatch.setattr(gw, '_served_by_another_host_gateway', lambda *a: owner)
    monkeypatch.setattr(gw, 'named_profile_served_by_running_multiplexer', lambda *a: True)
    calls = []

    def unserve(home, name):
        assert home == root and name == 'worker' and marker.exists()
        calls.append('unserve')
        return {'unserved': name, 'served_profiles': ['default']}

    def host_stop(*args, **kwargs):
        raise AssertionError('stop killed the host gateway pid')

    monkeypatch.setattr(control_socket, 'request_unserve_profile', unserve, raising=False)
    monkeypatch.setattr(gw, 'stop_profile_gateway', host_stop)
    gw._cmd_stop(SimpleNamespace())
    assert calls == ['unserve']
    assert marker.exists()
    assert "parked; its bots and cron are stopped" in capsys.readouterr().out


def test_separate_gateway_beside_the_host_keeps_its_own_lifecycle(homes, monkeypatch):
    from hermes_cli import gateway as gw
    from hermes_cli.gateway_profile_lifecycle import profile_lifecycle
    root, secondary = homes
    host_pid = 4242
    owner = SimpleNamespace(home=root, pid=host_pid, profile_label='default',
                            profiles=('default', 'worker'), describe=lambda: 'test host')
    monkeypatch.setenv('HERMES_HOME', str(secondary))
    monkeypatch.setattr(gw, '_current_profile_name', lambda: 'worker')
    monkeypatch.setattr(gw, 'find_gateway_pids', lambda **kw: [host_pid, 5151])
    monkeypatch.setattr(gw, 'host_multiplexer_serving', lambda *a: owner)
    assert profile_lifecycle('stop', SimpleNamespace()) is False


def test_parked_status_and_topology_keep_roster(homes, monkeypatch, capsys):
    from hermes_cli import gateway as gw, profiles
    from hermes_cli.web_server_gateway import _collect_profile_gateway_topology
    _root, secondary = homes
    (secondary / 'gateway.parked').touch()
    monkeypatch.setenv('HERMES_HOME', str(secondary))
    monkeypatch.setattr(gw, '_current_profile_name', lambda: 'worker')
    monkeypatch.setattr(profiles, 'get_active_profile_name', lambda: 'worker')
    monkeypatch.setattr(profiles, '_check_gateway_running', lambda home: False)
    gw._cmd_status(SimpleNamespace())
    assert 'parked (hermes -p worker gateway start)' in capsys.readouterr().out
    topology = _collect_profile_gateway_topology()
    assert topology['profiles'] == ['default', 'worker']
    assert topology['parked_profiles'] == ['worker']


def test_parked_profile_keeps_implicit_host_multiplexed(homes, monkeypatch):
    from gateway.config import GatewayConfig
    from hermes_cli import gateway_migrate
    from hermes_cli.gateway_multiplex_mode import resolve_multiplex_mode
    _, secondary = homes
    (secondary / 'gateway.parked').touch()
    monkeypatch.setattr(gateway_migrate, '_host_supports_migration', lambda: None)
    monkeypatch.setattr(gateway_migrate, 'build_migration_plan', lambda: SimpleNamespace(
        standalone_secondaries=[], blocked=False))
    config = GatewayConfig()
    assert resolve_multiplex_mode(config).enabled


def test_dashboard_exposes_parked_profile_and_start_unparks_it(homes, monkeypatch):
    from types import SimpleNamespace
    from fastapi.testclient import TestClient
    from gateway import host_attach
    from hermes_cli import web_server, profiles
    from hermes_cli.web_server_gateway import multiplexed_profile_refusal
    root, secondary = homes
    (secondary / 'gateway.parked').touch()
    monkeypatch.setattr(profiles, '_check_gateway_running', lambda home: False)
    with TestClient(web_server.app) as client:
        response = client.get('/api/status')
    assert response.status_code == 200
    assert response.json()['parked_profiles'] == ['worker']
    # The Start button: refused while no host can unpark it, allowed (spawns `-p worker gateway start`)
    # once the host multiplexer is live; a parked profile is not served, so this is not the served path.
    monkeypatch.setattr(host_attach, 'host_gateway', lambda: None)
    assert multiplexed_profile_refusal('worker', 'start')
    monkeypatch.setattr(host_attach, 'host_gateway', lambda: SimpleNamespace(home=root, pid=1))
    assert multiplexed_profile_refusal('worker', 'start') is None


@pytest.mark.parametrize('host_running', [False, True])
def test_start_unparks_without_host_rendezvous(homes, monkeypatch, capsys, host_running):
    from hermes_cli import gateway as gw, gateway_multiplex_served as served
    from gateway import control_socket
    root, secondary = homes
    marker = secondary / 'gateway.parked'
    marker.touch()
    monkeypatch.setenv('HERMES_HOME', str(secondary))
    monkeypatch.setattr(gw, '_current_profile_name', lambda: 'worker')
    monkeypatch.setattr(gw, '_host_multiplexer_for_all_verb', lambda: None)
    monkeypatch.setattr(served, 'live_default_gateway_pid', lambda: 42 if host_running else None)
    calls = []

    def serve(home, name):
        assert home == root and name == 'worker' and not marker.exists()
        calls.append('serve')
        return {'served': name}

    def normal_start(**kw):
        assert not marker.exists()
        calls.append('normal')

    monkeypatch.setattr(control_socket, 'request_serve_profile_hot', serve)
    monkeypatch.setattr(gw, '_guard_named_profile_under_multiplexer', normal_start)
    monkeypatch.setattr(gw, '_dispatch_via_service_manager_if_s6', lambda *a: True)
    gw._cmd_start(SimpleNamespace())
    assert not marker.exists()
    assert calls == (['serve'] if host_running else ['normal'])


def test_default_status_distinguishes_served_and_parked(homes, monkeypatch, capsys):
    from hermes_cli import gateway as gw
    from hermes_cli.gateway_profile_lifecycle import print_parked_status
    root, secondary = homes
    (secondary / 'gateway.parked').touch()
    monkeypatch.setattr(gw, '_current_profile_name', lambda: 'default')
    monkeypatch.setattr(gw, 'host_multiplexer_serving', lambda: SimpleNamespace(
        home=root, profiles=('default',)))
    assert print_parked_status() is False
    output = capsys.readouterr().out
    assert 'Served profiles: default' in output
    assert "Profile 'worker': parked" in output


def test_parked_status_still_reports_a_forced_gateway(homes, monkeypatch, capsys):
    from hermes_cli import gateway as gw, profiles
    _root, secondary = homes
    (secondary / 'gateway.parked').touch()
    monkeypatch.setenv('HERMES_HOME', str(secondary))
    monkeypatch.setattr(gw, '_current_profile_name', lambda: 'worker')
    monkeypatch.setattr(profiles, 'get_active_profile_name', lambda: 'worker')
    monkeypatch.setattr(gw, 'get_gateway_runtime_snapshot', lambda system=False: gw.GatewayRuntimeSnapshot(
        manager='manual', gateway_pids=(4242,)))
    monkeypatch.setattr(gw, 'named_profile_served_by_running_multiplexer', lambda: False)
    monkeypatch.setattr(gw, '_installed_service_kind_for', lambda _check: None)
    gw._cmd_status(SimpleNamespace())
    output = capsys.readouterr().out
    assert 'parked (hermes -p worker gateway start)' in output
    assert 'Gateway is running (PID: 4242)' in output


@pytest.mark.parametrize('verdict, healthy', [
    (('incompatible', 'runtime_protocol'), False),  # an older Hermes's daemon left running
    (('ready', None), True),
])
def test_status_flags_a_gateway_this_hermes_cannot_attach(homes, monkeypatch, capsys, verdict, healthy):
    """Upgrade with the old daemon up: every client refuses it, so status must not say ✓."""
    from hermes_cli import gateway as gw, gateway_runtime, profiles
    root, _secondary = homes
    monkeypatch.setenv('HERMES_HOME', str(root))
    monkeypatch.setattr(gw, '_current_profile_name', lambda: 'default')
    monkeypatch.setattr(profiles, 'get_active_profile_name', lambda: 'default')
    monkeypatch.setattr(gw, 'get_gateway_runtime_snapshot', lambda system=False: gw.GatewayRuntimeSnapshot(
        manager='manual', gateway_pids=(4242,)))
    monkeypatch.setattr(gw, 'named_profile_served_by_running_multiplexer', lambda: False)
    monkeypatch.setattr(gw, '_installed_service_kind_for', lambda _check: None)
    monkeypatch.setattr(gateway_runtime, 'discover_gateway_endpoint', lambda home, timeout=2.0: (
        gateway_runtime.GatewayDiscovery(verdict[0], reason_code=verdict[1])))
    gw._cmd_status(SimpleNamespace())
    output = capsys.readouterr().out
    assert ('✓ Gateway is running (PID: 4242)' in output) is healthy
    assert ('incompatible: runtime_protocol' in output) is not healthy
    assert ('hermes gateway restart' in output) is not healthy


@pytest.mark.parametrize('served', [True, False])
def test_process_discovery_excludes_host_even_after_unserve(homes, monkeypatch, served):
    from gateway import status, host_attach
    from hermes_cli import gateway as gw, dashboard_procs
    root, secondary = homes
    owner = SimpleNamespace(pid=4242, home=root, served_known=True, profiles=('default', 'worker') if served else ('default',))
    monkeypatch.setenv('HERMES_HOME', str(secondary))
    monkeypatch.setattr(host_attach, '_probe_host_gateway', lambda wait: owner)
    monkeypatch.setattr(dashboard_procs, '_hermes_home_for_pid', lambda pid, **kw: str(root if pid == 4242 else secondary))
    monkeypatch.setattr(gw, '_get_service_pids', lambda **kw: [])
    monkeypatch.setattr(gw, '_scan_gateway_pids', lambda *a, **kw: [])
    monkeypatch.setattr(status, 'get_running_pid', lambda **kw: owner.pid)
    assert gw.find_gateway_pids() == []
    assert gw.stop_profile_gateway() is False
    # A distinct --force gateway must retain its own lifecycle even while the host serves us.
    monkeypatch.setattr(status, 'get_running_pid', lambda **kw: 4343)
    assert gw.find_gateway_pids() == [4343]
    # The launching profile still owns the shared process.
    monkeypatch.setenv('HERMES_HOME', str(root))
    monkeypatch.setattr(status, 'get_running_pid', lambda **kw: owner.pid)
    assert gw.find_gateway_pids() == [owner.pid]


@pytest.mark.parametrize('probe', ['cached_absent', 'failure', 'switched'])
def test_stop_never_owns_shared_pid_from_an_absent_or_old_probe(homes, monkeypatch, probe):
    import time
    from gateway import host_attach, status
    from hermes_cli import gateway as gw, dashboard_procs
    root, secondary = homes
    monkeypatch.setenv('HERMES_HOME', str(secondary))
    monkeypatch.setattr(status, 'get_running_pid', lambda **kw: 4242)
    monkeypatch.setattr(gw, '_get_service_pids', lambda **kw: [])
    monkeypatch.setattr(gw, '_scan_gateway_pids', lambda *a, **kw: [])
    monkeypatch.setattr(dashboard_procs, '_hermes_home_for_pid', lambda pid, **kw: str(root))
    old = host_attach.HostGateway(4141, root, ('default', 'worker'))
    current = host_attach.HostGateway(4242, root, ('default', 'worker'))
    monkeypatch.setattr(host_attach, '_cached_probe', (time.monotonic(), old if probe == 'switched' else None))

    def fresh(_wait):
        if probe == 'failure':
            raise OSError('rendezvous unavailable')
        return current

    monkeypatch.setattr(host_attach, '_probe_host_gateway', fresh)
    monkeypatch.setattr(gw.os, 'kill', lambda *a: pytest.fail('shared host was signalled'))
    assert gw.find_gateway_pids() == []
    assert gw.stop_profile_gateway() is False


@pytest.mark.parametrize('process_home', [None, 'foreign', 'error'])
def test_absent_host_requires_positive_process_ownership(homes, monkeypatch, process_home):
    from gateway import host_attach, status
    from hermes_cli import gateway as gw, dashboard_procs
    root, secondary = homes
    monkeypatch.setenv('HERMES_HOME', str(secondary))
    monkeypatch.setattr(status, 'get_running_pid', lambda **kw: 4242)
    monkeypatch.setattr(host_attach, '_probe_host_gateway', lambda wait: None)

    def home(pid, **kw):
        if process_home == 'error':
            raise PermissionError('process inaccessible')
        return str(root) if process_home == 'foreign' else None

    monkeypatch.setattr(dashboard_procs, '_hermes_home_for_pid', home)
    monkeypatch.setattr(gw, '_reap_unsupervised_gateway_orphans', lambda: pytest.fail('unproven PID fell through to reaper'))
    monkeypatch.setattr(gw.os, 'kill', lambda *a: pytest.fail('unproven PID was signalled'))
    assert gw.stop_profile_gateway() is False


def test_owner_switch_during_probe_is_rejected(homes, monkeypatch):
    from gateway import host_attach, status
    from hermes_cli import gateway as gw, dashboard_procs
    _, secondary = homes
    monkeypatch.setenv('HERMES_HOME', str(secondary))
    pids = iter([4343, 4242])
    monkeypatch.setattr(status, 'get_running_pid', lambda **kw: next(pids))
    monkeypatch.setattr(host_attach, '_probe_host_gateway', lambda wait: None)
    monkeypatch.setattr(dashboard_procs, '_hermes_home_for_pid', lambda pid, **kw: str(secondary))
    assert gw._owned_gateway_pid() is None


@pytest.mark.parametrize('rendezvous', ['absent', 'unreadable'])
def test_sticky_profile_cannot_reassign_running_host(homes, monkeypatch, rendezvous):
    from gateway import host_attach, status
    from hermes_cli import dashboard_procs, gateway as gw, main_dashboard
    root, secondary = homes
    monkeypatch.setenv('HERMES_HOME', str(secondary))
    # The process started at the root, before the user's later sticky switch.
    monkeypatch.setattr(dashboard_procs, '_pid_environ', lambda pid: {'HERMES_HOME': str(root)})
    monkeypatch.setattr(main_dashboard, '_dashboard_cmdline_for_pid', lambda pid: ['hermes', 'gateway', 'run'])
    (root / 'active_profile').write_text('worker\n')
    assert dashboard_procs._hermes_home_for_pid(4242) == str(secondary)

    def probe(wait):
        if rendezvous == 'unreadable':
            raise PermissionError('rendezvous unreadable')

    monkeypatch.setattr(host_attach, '_probe_host_gateway', probe)
    monkeypatch.setattr(status, 'get_running_pid', lambda **kw: 4242)
    monkeypatch.setattr(gw, '_reap_unsupervised_gateway_orphans', lambda: pytest.fail('unproven PID reached reaper'))
    monkeypatch.setattr(gw.os, 'kill', lambda *a: pytest.fail('shared host was signalled'))
    assert gw._owned_gateway_pid() is None
    assert gw.stop_profile_gateway() is False


@pytest.mark.parametrize('identity', ['explicit', 'profile_home', 'supervised', 'ambiguous'])
def test_launch_identity_does_not_follow_sticky_selection(homes, monkeypatch, identity):
    from gateway import host_attach, status
    from hermes_cli import dashboard_procs, gateway as gw, main_dashboard
    root, secondary = homes
    (root / 'active_profile').write_text('other\n')
    env = {'HERMES_HOME': str(root)}
    argv = ['hermes', 'gateway', 'run']
    target = secondary
    if identity == 'explicit':
        argv[1:1] = ['-p', 'worker']
    elif identity == 'profile_home':
        env['HERMES_HOME'] = str(secondary)
    elif identity == 'supervised':
        env['HERMES_GATEWAY_EXTERNAL_SUPERVISOR'] = '1'
        target = root
    monkeypatch.setenv('HERMES_HOME', str(target))
    monkeypatch.setattr(dashboard_procs, '_pid_environ', lambda pid: env)
    monkeypatch.setattr(main_dashboard, '_dashboard_cmdline_for_pid', lambda pid: argv)
    monkeypatch.setattr(host_attach, '_probe_host_gateway', lambda wait: None)
    monkeypatch.setattr(status, 'get_running_pid', lambda **kw: 4242)
    assert gw._owned_gateway_pid() == (None if identity == 'ambiguous' else 4242)


@pytest.mark.parametrize('runtime', ['present', 'absent', 'unknown'])
def test_cli_stop_reports_unconfirmed_separately_from_absent(homes, monkeypatch, capsys, runtime):
    from gateway import status
    from hermes_cli import gateway as gw, gateway_profile_lifecycle as lifecycle
    monkeypatch.setattr(lifecycle, 'profile_lifecycle', lambda *a: False)
    monkeypatch.setattr(gw, '_refuse_from_inside_gateway', lambda *a: None)
    monkeypatch.setattr(gw, 'find_gateway_pids', lambda **kw: [])
    monkeypatch.setattr(gw, '_served_by_another_host_gateway', lambda: None)
    monkeypatch.setattr(gw, 'named_profile_served_by_running_multiplexer', lambda: False)
    monkeypatch.setattr(gw, '_dispatch_via_service_manager_if_s6', lambda *a: False)
    monkeypatch.setattr(gw, '_stop_installed_service', lambda *a: False)
    monkeypatch.setattr(gw, 'stop_profile_gateway', lambda: False)

    def pid():
        if runtime == 'unknown':
            raise OSError('cannot inspect gateway')
        return 4242 if runtime == 'present' else None

    monkeypatch.setattr(status, 'get_running_pid', pid)
    if runtime == 'absent':
        gw._cmd_stop(SimpleNamespace())
        assert 'No gateway running' in capsys.readouterr().out
    else:
        with pytest.raises(SystemExit) as result:
            gw._cmd_stop(SimpleNamespace())
        assert result.value.code != 0
        output = capsys.readouterr().out
        assert 'No gateway running' not in output
        assert 'not confirmed' in output


@pytest.mark.parametrize('verified,changed', [(True, False), (True, True), (False, False)])
def test_same_home_record_requires_identity_and_stable_pid(homes, monkeypatch, verified, changed):
    from gateway import host_attach, status
    from hermes_cli import gateway as gw, dashboard_procs
    root, _ = homes
    monkeypatch.setenv('HERMES_HOME', str(root))
    from gateway import host_rendezvous as hr
    owner = host_attach.HostGateway(4242, root, ('default',), served_known=verified)
    monkeypatch.setattr(host_attach, '_probe_host_gateway', lambda wait: owner)
    monkeypatch.setattr(hr, 'read_record', lambda role, **kw: SimpleNamespace(pid=4242, home=str(root)))
    monkeypatch.setattr(hr, 'record_names_live_owner', lambda rec: True)
    monkeypatch.setattr(dashboard_procs, '_hermes_home_for_pid', lambda pid, **kw: None)
    pids = iter([4242, 4343 if changed else 4242])
    monkeypatch.setattr(status, 'get_running_pid', lambda: next(pids))
    assert gw._owned_gateway_pid() == (4242 if verified and not changed else None)