"""Tests for hermes_cli.service_manager — the abstract ServiceManager
protocol, the detect_service_manager() entry point, and the host-side
adapter wrappers (Systemd / Launchd / Windows).

The s6 backend is added in Phase 3; its tests live alongside the
implementation in this same file once that phase ships.
"""
from __future__ import annotations

import pytest

from hermes_cli.service_manager import (
    LaunchdServiceManager,
    S6ServiceManager,
    ServiceManager,
    ServiceManagerKind,
    SystemdServiceManager,
    WindowsServiceManager,
    detect_service_manager,
    get_service_manager,
    validate_profile_name,
)


# ---------------------------------------------------------------------------
# validate_profile_name
# ---------------------------------------------------------------------------


def test_validate_profile_name_accepts_valid_names() -> None:
    # Smoke: known-good names should not raise.
    validate_profile_name("coder")
    validate_profile_name("my-profile")
    validate_profile_name("assistant_v2")
    validate_profile_name("a")
    validate_profile_name("0")
    validate_profile_name("0abc")


@pytest.mark.parametrize(
    "bad",
    [
        "",                  # empty
        "Coder",             # uppercase
        "foo/bar",           # path traversal
        "../escape",         # path traversal
        "-leading-dash",     # leading dash (s6 reads as a flag)
        "_leading_underscore",  # leading underscore
        "name with spaces",  # whitespace
        "name.with.dots",    # punctuation
        "a" * 252,           # too long
    ],
)
def test_validate_profile_name_rejects_invalid(bad: str) -> None:
    with pytest.raises(ValueError):
        validate_profile_name(bad)


# ---------------------------------------------------------------------------
# detect_service_manager
# ---------------------------------------------------------------------------


def test_detect_service_manager_returns_known_value() -> None:
    """Without mocking, the function must still return one of the
    advertised literals — anything else means a new platform branch
    was added without updating ServiceManagerKind."""
    result = detect_service_manager()
    assert result in ("systemd", "launchd", "windows", "s6", "none")


def test_detect_service_manager_s6_keys_off_s6_running_not_is_container(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """Regression: Fly runs s6-overlay as PID 1 in a Firecracker microVM, which
    is not a Docker/Podman container. Gating s6 detection on is_container() made
    the dispatch path inert on Fly, so `hermes gateway restart` spawned a
    foreground gateway that fought the supervised one. Detection must key off
    s6 being PID 1 (`_s6_running`) alone."""
    monkeypatch.setattr(
        "hermes_cli.service_manager._s6_running", lambda: True,
    )
    assert detect_service_manager() == "s6"


# ---------------------------------------------------------------------------
# _s6_running — must work for unprivileged users, not just root
# ---------------------------------------------------------------------------


def _patch_s6_paths(
    monkeypatch: pytest.MonkeyPatch,
    *,
    comm: str | OSError | None,
    basedir_is_dir: bool,
) -> None:
    """Stub /proc/1/comm and /run/s6/basedir for _s6_running tests."""
    from pathlib import Path as _Path

    real_read_text = _Path.read_text
    real_is_dir = _Path.is_dir

    def fake_read_text(self, *args, **kwargs):  # type: ignore[override]
        if str(self) == "/proc/1/comm":
            if isinstance(comm, OSError):
                raise comm
            if comm is None:
                raise FileNotFoundError(2, "No such file or directory")
            return comm + "\n"
        return real_read_text(self, *args, **kwargs)

    def fake_is_dir(self):  # type: ignore[override]
        if str(self) == "/run/s6/basedir":
            return basedir_is_dir
        return real_is_dir(self)

    monkeypatch.setattr(_Path, "read_text", fake_read_text)
    monkeypatch.setattr(_Path, "is_dir", fake_is_dir)


def test_s6_running_true_when_comm_and_basedir_match(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    from hermes_cli.service_manager import _s6_running

    _patch_s6_paths(monkeypatch, comm="s6-svscan", basedir_is_dir=True)
    assert _s6_running() is True


def test_s6_running_false_when_comm_is_wrong(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    from hermes_cli.service_manager import _s6_running

    # systemd as PID 1, basedir present from some stray s6 install
    _patch_s6_paths(monkeypatch, comm="systemd", basedir_is_dir=True)
    assert _s6_running() is False


def test_s6_running_false_when_basedir_missing(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    from hermes_cli.service_manager import _s6_running

    # The comm matches but the basedir is missing — e.g. an unrelated
    # process happens to be named "s6-svscan"
    _patch_s6_paths(monkeypatch, comm="s6-svscan", basedir_is_dir=False)
    assert _s6_running() is False


def test_s6_running_false_when_comm_unreadable(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """Regression: /proc/1/exe was unreadable to UID 10000 and
    resolve() silently returned the unresolved path, making detection
    always-False inside the container under the hermes user. The new
    probe must FAIL CLOSED — not raise — when /proc/1/comm can't be
    read.
    """
    from hermes_cli.service_manager import _s6_running

    _patch_s6_paths(
        monkeypatch,
        comm=PermissionError(13, "Permission denied"),
        basedir_is_dir=True,
    )
    assert _s6_running() is False


def test_s6_running_handles_missing_proc(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """On macOS / Windows / WSL-without-procfs, /proc/1/comm doesn't
    exist. Must return False, not raise."""
    from hermes_cli.service_manager import _s6_running

    _patch_s6_paths(monkeypatch, comm=None, basedir_is_dir=False)
    assert _s6_running() is False


# ---------------------------------------------------------------------------
# Backend wrappers — kind + registration unsupported on hosts
# ---------------------------------------------------------------------------


def test_systemd_manager_kind_and_registration_unsupported() -> None:
    mgr = SystemdServiceManager()
    assert mgr.kind == "systemd"
    assert mgr.supports_runtime_registration() is False
    with pytest.raises(NotImplementedError):
        mgr.register_profile_gateway("foo")
    with pytest.raises(NotImplementedError):
        mgr.unregister_profile_gateway("foo")
    assert mgr.list_profile_gateways() == []
    # Protocol conformance — runtime_checkable lets us assert this.
    assert isinstance(mgr, ServiceManager)


def test_launchd_manager_kind_and_registration_unsupported() -> None:
    mgr = LaunchdServiceManager()
    assert mgr.kind == "launchd"
    assert mgr.supports_runtime_registration() is False
    with pytest.raises(NotImplementedError):
        mgr.register_profile_gateway("foo")
    assert mgr.list_profile_gateways() == []
    assert isinstance(mgr, ServiceManager)


def test_windows_manager_kind_and_registration_unsupported() -> None:
    mgr = WindowsServiceManager()
    assert mgr.kind == "windows"
    assert mgr.supports_runtime_registration() is False
    with pytest.raises(NotImplementedError):
        mgr.register_profile_gateway("foo")
    assert isinstance(mgr, ServiceManager)


# ---------------------------------------------------------------------------
# Lifecycle delegation — wrappers must call through to module-level fns
# ---------------------------------------------------------------------------


def test_systemd_manager_lifecycle_delegates(monkeypatch: pytest.MonkeyPatch) -> None:
    called: list[str] = []
    monkeypatch.setattr(
        "hermes_cli.gateway.systemd_start", lambda: called.append("start"),
    )
    monkeypatch.setattr(
        "hermes_cli.gateway.systemd_stop", lambda: called.append("stop"),
    )
    monkeypatch.setattr(
        "hermes_cli.gateway.systemd_restart", lambda: called.append("restart"),
    )
    monkeypatch.setattr(
        "hermes_cli.gateway._probe_systemd_service_running",
        lambda *a, **kw: (False, True),
    )
    mgr = SystemdServiceManager()
    mgr.start("ignored")
    mgr.stop("ignored")
    mgr.restart("ignored")
    assert called == ["start", "stop", "restart"]
    assert mgr.is_running("ignored") is True


def test_launchd_manager_lifecycle_delegates(monkeypatch: pytest.MonkeyPatch) -> None:
    called: list[str] = []
    monkeypatch.setattr(
        "hermes_cli.gateway.launchd_start", lambda: called.append("start"),
    )
    monkeypatch.setattr(
        "hermes_cli.gateway.launchd_stop", lambda: called.append("stop"),
    )
    monkeypatch.setattr(
        "hermes_cli.gateway.launchd_restart", lambda: called.append("restart"),
    )
    monkeypatch.setattr(
        "hermes_cli.gateway._probe_launchd_service_running", lambda: False,
    )
    mgr = LaunchdServiceManager()
    mgr.start("ignored")
    mgr.stop("ignored")
    mgr.restart("ignored")
    assert called == ["start", "stop", "restart"]
    assert mgr.is_running("ignored") is False


def test_windows_manager_lifecycle_delegates(monkeypatch: pytest.MonkeyPatch) -> None:
    called: list[str] = []
    # Force-import the submodule so monkeypatch's attribute lookup
    # against the `hermes_cli` package succeeds — gateway_windows is
    # imported lazily inside the wrapper and may not yet be loaded.
    import hermes_cli.gateway_windows  # noqa: F401

    class _FakeWindowsModule:
        @staticmethod
        def start() -> None: called.append("start")
        @staticmethod
        def stop() -> None: called.append("stop")
        @staticmethod
        def restart() -> None: called.append("restart")
        @staticmethod
        def is_installed() -> bool: return True

    monkeypatch.setattr("hermes_cli.gateway_windows", _FakeWindowsModule)
    monkeypatch.setattr(
        "hermes_cli.gateway.find_gateway_pids",
        lambda **kw: [12345],
    )
    mgr = WindowsServiceManager()
    mgr.start("ignored")
    mgr.stop("ignored")
    mgr.restart("ignored")
    assert called == ["start", "stop", "restart"]
    assert mgr.is_running("ignored") is True


def test_windows_manager_is_running_false_when_not_installed(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    import hermes_cli.gateway_windows  # noqa: F401

    class _FakeWindowsModule:
        @staticmethod
        def is_installed() -> bool: return False

    monkeypatch.setattr("hermes_cli.gateway_windows", _FakeWindowsModule)
    monkeypatch.setattr(
        "hermes_cli.gateway.find_gateway_pids",
        lambda **kw: [12345],  # PIDs would otherwise vote "running"
    )
    assert WindowsServiceManager().is_running("ignored") is False


def test_windows_manager_install_forwards_kwargs(monkeypatch: pytest.MonkeyPatch) -> None:
    captured: dict[str, object] = {}
    import hermes_cli.gateway_windows  # noqa: F401

    class _FakeWindowsModule:
        @staticmethod
        def install(*, force, start_now, start_on_login, elevated_handoff) -> None:
            captured["force"] = force
            captured["start_now"] = start_now
            captured["start_on_login"] = start_on_login
            captured["elevated_handoff"] = elevated_handoff

    monkeypatch.setattr("hermes_cli.gateway_windows", _FakeWindowsModule)
    WindowsServiceManager().install(
        force=True, start_now=True, start_on_login=False, elevated_handoff=True,
    )
    assert captured == {
        "force": True,
        "start_now": True,
        "start_on_login": False,
        "elevated_handoff": True,
    }


# ---------------------------------------------------------------------------
# get_service_manager factory
# ---------------------------------------------------------------------------


@pytest.mark.parametrize(
    "kind,cls",
    [
        ("systemd", SystemdServiceManager),
        ("launchd", LaunchdServiceManager),
        ("windows", WindowsServiceManager),
    ],
)
def test_get_service_manager_returns_correct_backend(
    monkeypatch: pytest.MonkeyPatch,
    kind: ServiceManagerKind,
    cls: type,
) -> None:
    monkeypatch.setattr(
        "hermes_cli.service_manager.detect_service_manager", lambda: kind,
    )
    assert isinstance(get_service_manager(), cls)


def test_get_service_manager_raises_when_unsupported(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    monkeypatch.setattr(
        "hermes_cli.service_manager.detect_service_manager", lambda: "none",
    )
    with pytest.raises(RuntimeError, match="no supported service manager"):
        get_service_manager()


def test_get_service_manager_returns_s6_instance(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """The s6 backend ships in Phase 3 — the factory must return an
    S6ServiceManager when running inside a container."""
    monkeypatch.setattr(
        "hermes_cli.service_manager.detect_service_manager", lambda: "s6",
    )
    assert isinstance(get_service_manager(), S6ServiceManager)


# ---------------------------------------------------------------------------
# S6ServiceManager — unit tests against a tmp-path scandir (no real s6)
# ---------------------------------------------------------------------------


@pytest.fixture
def s6_scandir(tmp_path):
    """Empty scandir for the S6ServiceManager tests."""
    d = tmp_path / "service"
    d.mkdir()
    return d


@pytest.fixture
def fake_subprocess_run(monkeypatch: pytest.MonkeyPatch):
    """Capture subprocess.run calls + always return success. Lets the
    S6ServiceManager tests run on hosts that don't have s6-svc /
    s6-svscanctl installed.

    Records are normalized: leading ``/command/`` is stripped from
    cmd[0] so assertions can match on the bare s6-svc / s6-svstat /
    s6-svscanctl name regardless of whether the manager calls them
    via absolute path or bare name."""
    calls: list[list[str]] = []

    def _fake(cmd, **kw):
        import subprocess as _sp
        seq = list(cmd) if isinstance(cmd, (list, tuple)) else [str(cmd)]
        if seq and seq[0].startswith("/command/"):
            seq[0] = seq[0][len("/command/"):]
        calls.append(seq)
        return _sp.CompletedProcess(cmd, 0, "", "")

    monkeypatch.setattr("subprocess.run", _fake)
    return calls


def test_s6_manager_kind_and_supports_registration() -> None:
    mgr = S6ServiceManager()
    assert mgr.kind == "s6"
    assert mgr.supports_runtime_registration() is True


# ---------------------------------------------------------------------------
# _seed_supervise_skeleton — unit tests
# ---------------------------------------------------------------------------
#
# The skeleton helper pre-creates the dirs and FIFOs that s6-supervise
# would otherwise create as root mode 0700, locking out the
# unprivileged hermes user from every lifecycle op. These tests run
# against tmp_path and assert the produced layout — the live-container
# verification (against real s6-svc / s6-svstat) lives in
# tests/docker/test_s6_profile_gateway_integration.py.


def test_seed_supervise_skeleton_creates_expected_layout(tmp_path) -> None:
    """Verifies the dirs + FIFO + modes the helper lays down."""
    import stat

    from hermes_cli.service_manager import _seed_supervise_skeleton

    svc_dir = tmp_path / "gateway-foo"
    svc_dir.mkdir()

    _seed_supervise_skeleton(svc_dir)

    # Top-level event/ — s6-svlisten1 event subscription dir.
    event = svc_dir / "event"
    assert event.is_dir(), "missing top-level event/"
    assert stat.S_IMODE(event.stat().st_mode) == 0o3730, (
        f"event/ mode = {oct(event.stat().st_mode)}, want 03730"
    )

    # supervise/ dir.
    supervise = svc_dir / "supervise"
    assert supervise.is_dir(), "missing supervise/"
    assert stat.S_IMODE(supervise.stat().st_mode) == 0o755

    # supervise/event/.
    supervise_event = supervise / "event"
    assert supervise_event.is_dir(), "missing supervise/event/"
    assert stat.S_IMODE(supervise_event.stat().st_mode) == 0o3730

    # supervise/control FIFO.
    control = supervise / "control"
    assert control.exists(), "missing supervise/control FIFO"
    assert stat.S_ISFIFO(control.stat().st_mode), (
        "supervise/control must be a FIFO"
    )
    assert stat.S_IMODE(control.stat().st_mode) == 0o660


def test_seed_supervise_skeleton_handles_log_subservice(tmp_path) -> None:
    """When a log/ subdir exists, its supervise tree also gets seeded.

    Without this, ``unregister_profile_gateway``'s rmtree would EACCES
    on the logger's root-owned supervise dir even after the parent
    slot's supervise/ was hermes-owned.
    """
    import stat

    from hermes_cli.service_manager import _seed_supervise_skeleton

    svc_dir = tmp_path / "gateway-foo"
    svc_dir.mkdir()
    (svc_dir / "log").mkdir()  # logger subdir present

    _seed_supervise_skeleton(svc_dir)

    # Logger's own supervise tree is seeded the same way.
    log_event = svc_dir / "log" / "event"
    log_supervise = svc_dir / "log" / "supervise"
    log_supervise_event = log_supervise / "event"
    log_control = log_supervise / "control"

    assert log_event.is_dir()
    assert stat.S_IMODE(log_event.stat().st_mode) == 0o3730
    assert log_supervise.is_dir()
    assert log_supervise_event.is_dir()
    assert log_control.exists() and stat.S_ISFIFO(log_control.stat().st_mode)


def test_seed_supervise_skeleton_skips_when_no_log_subservice(tmp_path) -> None:
    """If log/ isn't present, no logger skeleton is created."""
    from hermes_cli.service_manager import _seed_supervise_skeleton

    svc_dir = tmp_path / "gateway-foo"
    svc_dir.mkdir()

    _seed_supervise_skeleton(svc_dir)

    assert not (svc_dir / "log").exists(), (
        "helper must not synthesize a log/ subdir on its own"
    )


def test_seed_supervise_skeleton_is_idempotent(tmp_path) -> None:
    """Calling the helper twice on the same dir is a no-op the second time.

    Important because s6-supervise may have already opened the FIFO
    when a re-register / reconcile happens; double-creation would
    error out. The helper short-circuits on existence.
    """
    from hermes_cli.service_manager import _seed_supervise_skeleton

    svc_dir = tmp_path / "gateway-foo"
    svc_dir.mkdir()

    _seed_supervise_skeleton(svc_dir)
    _seed_supervise_skeleton(svc_dir)  # must not raise


def test_s6_register_creates_service_dir_and_triggers_scan(
    s6_scandir, fake_subprocess_run,
) -> None:
    mgr = S6ServiceManager(scandir=s6_scandir)
    mgr.register_profile_gateway("coder")

    svc_dir = s6_scandir / "gateway-coder"
    assert svc_dir.is_dir()
    assert (svc_dir / "type").read_text().strip() == "longrun"

    run_path = svc_dir / "run"
    assert run_path.is_file()
    assert run_path.stat().st_mode & 0o111  # executable
    run_text = run_path.read_text()
    assert "export HOME=/opt/data" in run_text
    assert "hermes -p coder gateway run" in run_text
    assert "s6-setuidgid hermes" in run_text
    # Sentinel marking this as the supervised-child invocation. Without
    # it, the supervised `gateway run` would re-enter the s6 redirect
    # in `_gateway_command_inner` and recurse. See the matching guard
    # in hermes_cli/gateway.py::_gateway_command_inner.
    assert "export HERMES_S6_SUPERVISED_CHILD=1" in run_text

    log_run = svc_dir / "log" / "run"
    assert log_run.is_file()
    log_text = log_run.read_text()
    # CRITICAL: HERMES_HOME must be a runtime env-var expansion, NOT
    # a Python-substituted absolute path. Negative-assert the wrong
    # form so future regressions are caught.
    assert "$HERMES_HOME" in log_text
    assert "logs/gateways/coder" in log_text
    assert "/opt/data/logs/gateways/coder" not in log_text, (
        "log_dir was hard-coded; must use ${HERMES_HOME} at run time"
    )
    # `1` action directive forwards lines to stdout BEFORE the file
    # destination so the supervised gateway's stdout (including the
    # rich-console banner and plain print() output) reaches docker
    # logs, not just the rotated file. See _render_log_run's docstring
    # for the full output-routing rationale.
    assert "s6-log 1 " in log_text, (
        "log/run must include the `1` action directive before the file "
        "destination so supervised stdout reaches docker logs. Saw: "
        f"{log_text!r}"
    )

    # s6-svscanctl -a was invoked against the scandir
    assert any(
        cmd[0] == "s6-svscanctl" and "-a" in cmd
        and str(s6_scandir) in cmd
        for cmd in fake_subprocess_run
    ), f"s6-svscanctl -a not invoked; saw: {fake_subprocess_run}"


def test_s6_register_staging_dir_is_dotfile_hidden_from_svscan(
    s6_scandir, fake_subprocess_run, monkeypatch: pytest.MonkeyPatch,
) -> None:
    """The mid-build staging dir MUST be dot-prefixed so s6-svscan
    ignores it while it is half-populated.

    s6-svscan skips any scandir entry whose name begins with ``.``. If
    the staging dir were a plain ``gateway-<p>.tmp`` (a non-dotfile),
    a concurrent ``s6-svscanctl -a`` rescan would supervise it the
    moment it has a valid ``type``/``run`` — spawning s6-supervise AS
    ROOT, which mkdir's a root-owned ``supervise/`` and makes the
    in-flight ``_seed_supervise_skeleton`` EACCES on
    ``mkdir supervise/event``. That was the arm64-only CI flake on
    ``test_s6_unregister_removes_service_dir_in_live_container``.

    We capture the directory passed to ``_seed_supervise_skeleton``
    (called mid-build, BEFORE the atomic rename to the live name) and
    assert its basename starts with ``.`` and still lives in the
    scandir as a sibling of the live slot.
    """
    import hermes_cli.service_manager as sm

    seen: list[str] = []
    real_seed = sm._seed_supervise_skeleton

    def _capturing_seed(svc_dir, *a, **kw):
        seen.append(str(svc_dir))
        return real_seed(svc_dir, *a, **kw)

    monkeypatch.setattr(sm, "_seed_supervise_skeleton", _capturing_seed)

    S6ServiceManager(scandir=s6_scandir).register_profile_gateway("coder")

    assert seen, "_seed_supervise_skeleton was never called during register"
    staging = seen[0]
    staging_name = staging.rsplit("/", 1)[-1]
    assert staging_name.startswith("."), (
        f"staging dir must be a dotfile so s6-svscan skips it mid-build; "
        f"got {staging_name!r}"
    )
    # Sibling of the live slot, in the same scandir.
    assert staging == str(s6_scandir / ".gateway-coder.tmp")
    # And the published (renamed) live slot is the dotless canonical name.
    assert (s6_scandir / "gateway-coder").is_dir()


def test_s6_register_start_now_false_writes_down_marker(
    s6_scandir, fake_subprocess_run,
) -> None:
    """When start_now=False, a `down` marker must be written so
    s6-supervise does not auto-start the service on rescan."""
    mgr = S6ServiceManager(scandir=s6_scandir)
    mgr.register_profile_gateway("coder", start_now=False)

    svc_dir = s6_scandir / "gateway-coder"
    assert svc_dir.is_dir()
    assert (svc_dir / "down").is_file(), (
        "start_now=False must write a `down` marker file"
    )


def test_s6_register_start_now_true_no_down_marker(
    s6_scandir, fake_subprocess_run,
) -> None:
    """When start_now=True (default), no `down` marker should exist."""
    mgr = S6ServiceManager(scandir=s6_scandir)
    mgr.register_profile_gateway("coder")

    svc_dir = s6_scandir / "gateway-coder"
    assert svc_dir.is_dir()
    assert not (svc_dir / "down").exists(), (
        "start_now=True must NOT write a `down` marker file"
    )


def test_s6_register_extra_env_is_quoted(s6_scandir, fake_subprocess_run) -> None:
    mgr = S6ServiceManager(scandir=s6_scandir)
    mgr.register_profile_gateway(
        "x", extra_env={"FOO": "bar baz", "QUOTED": "a'b"},
    )
    run_text = (s6_scandir / "gateway-x" / "run").read_text()
    # shlex.quote should have wrapped both values
    assert "export FOO='bar baz'" in run_text
    assert "export QUOTED='a'\"'\"'b'" in run_text


def test_render_run_script_resets_home_before_exec() -> None:

    run_text = S6ServiceManager._render_run_script("coder", {})

    assert "export HOME=/opt/data" in run_text
    assert "exec s6-setuidgid hermes hermes -p coder gateway run --replace" in run_text


def test_render_run_script_uses_replace_to_take_over_stale_holder() -> None:
    """NS-505: the supervised gateway must exec ``gateway run --replace``.

    Without ``--replace`` a gateway started OUTSIDE s6 (a stray shell
    ``hermes gateway run``, an agent action, the Open WebUI helper) holds
    the per-HERMES_HOME PID lock; the supervised slot then execs a bare
    ``gateway run``, hits the "Another gateway instance is already
    running" guard, exits non-zero, and s6 restarts it — a restart loop
    that never binds. ``--replace`` makes the supervised gateway reap the
    stale holder and win, so s6 is authoritative for the slot.

    Covers both the default (root HERMES_HOME, no ``-p``) and named-profile
    render paths.
    """
    default_text = S6ServiceManager._render_run_script("default", {})
    # Root profile: bare `hermes gateway run --replace` (no -p flag).
    assert "hermes gateway run --replace" in default_text
    assert "hermes -p default" not in default_text
    # Every exec line that launches the gateway must carry --replace, so
    # neither the non-root nor the privilege-drop branch can spin.
    gateway_execs = [
        line for line in default_text.splitlines()
        if "gateway run" in line
    ]
    assert gateway_execs, "no gateway run exec line rendered"
    assert all("--replace" in line for line in gateway_execs), (
        f"a gateway run line is missing --replace: {gateway_execs}"
    )

    named_text = S6ServiceManager._render_run_script("coder", {})
    named_execs = [
        line for line in named_text.splitlines() if "gateway run" in line
    ]
    assert named_execs
    assert all("--replace" in line for line in named_execs), (
        f"a named-profile gateway run line is missing --replace: {named_execs}"
    )


def test_render_finish_script_exits_125_on_ex_config() -> None:
    """The finish script must translate exit 78 (EX_CONFIG) into exit 125
    (permanent failure) so s6 stops restarting on fatal config errors.
    See #51228."""
    text = S6ServiceManager._render_finish_script()
    assert '[ "$1" = "78" ]' in text
    assert "exit 125" in text
    assert "exit 0" in text


def test_s6_register_writes_finish_script(
    s6_scandir, fake_subprocess_run,
) -> None:
    """The finish script must be written alongside the run script."""
    mgr = S6ServiceManager(scandir=s6_scandir)
    mgr.register_profile_gateway("coder")

    finish_path = s6_scandir / "gateway-coder" / "finish"
    assert finish_path.is_file()
    assert finish_path.stat().st_mode & 0o111  # executable
    assert "78" in finish_path.read_text()
    assert "125" in finish_path.read_text()


def test_s6_register_rejects_invalid_profile_name(s6_scandir) -> None:
    mgr = S6ServiceManager(scandir=s6_scandir)
    with pytest.raises(ValueError):
        mgr.register_profile_gateway("Bad/Name")


def test_s6_register_rejects_duplicate(s6_scandir, fake_subprocess_run) -> None:
    mgr = S6ServiceManager(scandir=s6_scandir)
    (s6_scandir / "gateway-coder").mkdir(parents=True)
    with pytest.raises(ValueError, match="already registered"):
        mgr.register_profile_gateway("coder")


def test_s6_register_rolls_back_on_svscanctl_failure(
    s6_scandir, monkeypatch: pytest.MonkeyPatch,
) -> None:
    """If s6-svscanctl fails the service dir must be cleaned up so the
    next register call doesn't see a stale duplicate."""
    import subprocess as _sp

    def _fail_scanctl(cmd, **kw):
        # Manager calls s6-svscanctl by absolute path; match on basename.
        if cmd[0].endswith("/s6-svscanctl"):
            return _sp.CompletedProcess(cmd, 1, "", "rescan failed")
        return _sp.CompletedProcess(cmd, 0, "", "")
    monkeypatch.setattr("subprocess.run", _fail_scanctl)

    mgr = S6ServiceManager(scandir=s6_scandir)
    with pytest.raises(RuntimeError, match="s6-svscanctl failed"):
        mgr.register_profile_gateway("coder")
    assert not (s6_scandir / "gateway-coder").exists()


def test_s6_unregister_removes_service_dir(
    s6_scandir, fake_subprocess_run,
) -> None:
    svc_dir = s6_scandir / "gateway-coder"
    svc_dir.mkdir(parents=True)
    (svc_dir / "type").write_text("longrun\n")

    mgr = S6ServiceManager(scandir=s6_scandir)
    mgr.unregister_profile_gateway("coder")

    # s6-svc -d was issued
    assert any(
        cmd[0] == "s6-svc" and "-d" in cmd
        for cmd in fake_subprocess_run
    )
    # Service dir was removed
    assert not svc_dir.exists()
    # Rescan was triggered
    assert any(cmd[0] == "s6-svscanctl" for cmd in fake_subprocess_run)


def test_s6_unregister_absent_profile_is_noop(s6_scandir) -> None:
    # Should NOT raise even though "ghost" doesn't exist
    S6ServiceManager(scandir=s6_scandir).unregister_profile_gateway("ghost")


def test_s6_list_profile_gateways(s6_scandir) -> None:
    # Three gateway profiles + one unrelated service + one hidden dir
    (s6_scandir / "gateway-coder").mkdir()
    (s6_scandir / "gateway-assistant").mkdir()
    (s6_scandir / "gateway-writer").mkdir()
    (s6_scandir / "s6-linux-init-shutdownd").mkdir()  # filtered out
    (s6_scandir / ".lock").mkdir()  # filtered out (hidden)

    profiles = sorted(S6ServiceManager(scandir=s6_scandir).list_profile_gateways())
    assert profiles == ["assistant", "coder", "writer"]


def test_s6_list_profile_gateways_empty_when_scandir_missing(tmp_path) -> None:
    missing = tmp_path / "does-not-exist"
    assert S6ServiceManager(scandir=missing).list_profile_gateways() == []


def test_s6_lifecycle_dispatches_to_s6_svc(
    s6_scandir, fake_subprocess_run,
) -> None:
    mgr = S6ServiceManager(scandir=s6_scandir)
    # _run_svc now verifies the slot exists before invoking s6-svc, so
    # we have to pre-seed the dir. In real use the slot is created by
    # register_profile_gateway or the cont-init.d reconciler.
    (s6_scandir / "gateway-coder").mkdir()
    mgr.start("gateway-coder")
    mgr.stop("gateway-coder")
    mgr.restart("gateway-coder")

    flags = [c[1] for c in fake_subprocess_run if c[0] == "s6-svc"]
    assert flags == ["-u", "-d", "-t"]


def test_s6_lifecycle_persists_named_profile_desired_state(
    s6_scandir,
    fake_subprocess_run,
    tmp_path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    import json

    hermes_home = tmp_path / "hermes-home"
    profile_dir = hermes_home / "profiles" / "coder"
    profile_dir.mkdir(parents=True)
    (s6_scandir / "gateway-coder").mkdir()
    monkeypatch.setenv("HERMES_HOME", str(hermes_home))

    mgr = S6ServiceManager(scandir=s6_scandir)
    mgr.start("gateway-coder")
    assert json.loads((profile_dir / "gateway_state.json").read_text())["desired_state"] == "running"
    mgr.stop("gateway-coder")
    assert json.loads((profile_dir / "gateway_state.json").read_text())["desired_state"] == "stopped"
    mgr.restart("gateway-coder")
    assert json.loads((profile_dir / "gateway_state.json").read_text())["desired_state"] == "running"


def test_s6_lifecycle_persists_default_profile_desired_state(
    s6_scandir,
    fake_subprocess_run,
    tmp_path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    import json

    hermes_home = tmp_path / "hermes-home"
    hermes_home.mkdir()
    (s6_scandir / "gateway-default").mkdir()
    monkeypatch.setenv("HERMES_HOME", str(hermes_home / "profiles" / "coder"))

    mgr = S6ServiceManager(scandir=s6_scandir)
    mgr.start("gateway-default")
    state = json.loads((hermes_home / "gateway_state.json").read_text())
    assert state["desired_state"] == "running"


# ---------------------------------------------------------------------------
# Lifecycle errors — friendly messages, not raw CalledProcessError
# ---------------------------------------------------------------------------


def test_lifecycle_raises_gateway_not_registered_for_missing_slot(
    s6_scandir, fake_subprocess_run,
) -> None:
    """When the service slot doesn't exist, the lifecycle methods
    must raise GatewayNotRegisteredError BEFORE invoking s6-svc, so
    the user sees a clear 'no such gateway' message instead of an
    opaque CalledProcessError stacktrace."""
    from hermes_cli.service_manager import (
        GatewayNotRegisteredError,
    )

    mgr = S6ServiceManager(scandir=s6_scandir)
    # No gateway-typo/ directory exists — slot is missing.
    with pytest.raises(GatewayNotRegisteredError) as excinfo:
        mgr.start("gateway-typo")
    assert excinfo.value.profile == "typo"
    assert excinfo.value.service == "gateway-typo"
    msg = str(excinfo.value)
    assert "'typo'" in msg
    assert "hermes profile create typo" in msg
    # And critically: s6-svc was NOT invoked.
    assert not any(c[0] == "s6-svc" for c in fake_subprocess_run)


@pytest.mark.parametrize("action,method_name", [
    ("start", "start"),
    ("stop", "stop"),
    ("restart", "restart"),
])
def test_all_lifecycle_methods_check_for_missing_slot(
    s6_scandir,
    fake_subprocess_run,
    action: str,
    method_name: str,
) -> None:
    """start/stop/restart all check for missing slots the same way."""
    from hermes_cli.service_manager import (
        GatewayNotRegisteredError,
    )

    mgr = S6ServiceManager(scandir=s6_scandir)
    with pytest.raises(GatewayNotRegisteredError):
        getattr(mgr, method_name)("gateway-absent")


def test_gateway_not_registered_unprefixed_service_name(s6_scandir) -> None:
    """If the caller passes a name without the 'gateway-' prefix (the
    Protocol allows arbitrary service names), the error still carries
    that name verbatim as the 'profile' so error messages don't
    accidentally strip user-provided text."""
    from hermes_cli.service_manager import (
        GatewayNotRegisteredError,
    )

    mgr = S6ServiceManager(scandir=s6_scandir)
    with pytest.raises(GatewayNotRegisteredError) as excinfo:
        mgr.start("not-prefixed")
    assert excinfo.value.profile == "not-prefixed"


def test_lifecycle_raises_s6_command_error_on_subprocess_failure(
    s6_scandir, monkeypatch: pytest.MonkeyPatch,
) -> None:
    """When s6-svc itself fails (non-zero exit) — e.g. EACCES on the
    supervise control FIFO — the lifecycle methods translate the
    CalledProcessError into a named S6CommandError carrying the
    return code and stderr."""
    import subprocess as _sp
    from hermes_cli.service_manager import S6CommandError

    # Pre-create the slot so we reach the s6-svc call.
    (s6_scandir / "gateway-coder").mkdir()

    def _fail(cmd, **kw):
        raise _sp.CalledProcessError(
            returncode=111,
            cmd=cmd,
            stderr="s6-svc: fatal: unable to control supervise/control: "
                   "Permission denied\n",
        )
    monkeypatch.setattr("subprocess.run", _fail)

    mgr = S6ServiceManager(scandir=s6_scandir)
    with pytest.raises(S6CommandError) as excinfo:
        mgr.start("gateway-coder")
    assert excinfo.value.service == "gateway-coder"
    assert excinfo.value.action == "start"
    assert excinfo.value.returncode == 111
    assert "Permission denied" in excinfo.value.stderr
    assert "Permission denied" in str(excinfo.value)
    assert "rc=111" in str(excinfo.value)


def test_s6_is_running_parses_svstat(
    s6_scandir, monkeypatch: pytest.MonkeyPatch,
) -> None:
    import subprocess as _sp

    def _svstat(cmd, **kw):
        if cmd[0].endswith("/s6-svstat"):
            return _sp.CompletedProcess(cmd, 0, "up (pid 42) 17 seconds\n", "")
        return _sp.CompletedProcess(cmd, 0, "", "")
    monkeypatch.setattr("subprocess.run", _svstat)
    assert S6ServiceManager(scandir=s6_scandir).is_running("gateway-coder") is True

    def _svstat_down(cmd, **kw):
        if cmd[0].endswith("/s6-svstat"):
            return _sp.CompletedProcess(cmd, 0, "down 5 seconds\n", "")
        return _sp.CompletedProcess(cmd, 0, "", "")
    monkeypatch.setattr("subprocess.run", _svstat_down)
    assert S6ServiceManager(scandir=s6_scandir).is_running("gateway-coder") is False


# ---------------------------------------------------------------------------
# S6 stop writes a planned-stop marker (issue #42675)
#
# `hermes gateway stop` inside a container dispatches through
# S6ServiceManager.stop() -> `s6-svc -d`, which SIGTERMs the gateway.
# That SIGTERM is indistinguishable from the one s6/Docker sends on a
# container restart unless we mark the intentional stop first. Without
# the marker, the gateway's shutdown handler can't tell an operator
# stop from a restart kill, and the gateway_state=stopped suppression
# (run.py) would never engage for explicit stops.
# ---------------------------------------------------------------------------


def test_s6_supervised_pid_parses_svstat(monkeypatch, s6_scandir):
    """_supervised_pid extracts the PID from `up (pid NNNN) ...`."""
    import subprocess as _sp

    def _fake(cmd, **kw):
        return _sp.CompletedProcess(cmd, 0, "up (pid 4242) 17 seconds\n", "")

    monkeypatch.setattr("subprocess.run", _fake)
    mgr = S6ServiceManager(scandir=s6_scandir)
    assert mgr._supervised_pid("gateway-coder") == 4242


def test_s6_supervised_pid_none_when_down(monkeypatch, s6_scandir):
    """A down service (`s6-svstat` rc!=0 or no pid) yields None."""
    import subprocess as _sp

    def _fake(cmd, **kw):
        return _sp.CompletedProcess(cmd, 0, "down (exitcode 0) 3 seconds\n", "")

    monkeypatch.setattr("subprocess.run", _fake)
    mgr = S6ServiceManager(scandir=s6_scandir)
    assert mgr._supervised_pid("gateway-coder") is None


def test_s6_stop_writes_planned_stop_marker(monkeypatch, s6_scandir):
    """stop() must mark the supervised PID before `s6-svc -d` so the
    gateway recognises the SIGTERM as an intentional stop (#42675)."""
    import subprocess as _sp

    svc_dir = s6_scandir / "gateway-coder"
    svc_dir.mkdir()  # so _run_svc doesn't raise GatewayNotRegisteredError

    svc_calls: list[list[str]] = []

    def _fake(cmd, **kw):
        seq = list(cmd) if isinstance(cmd, (list, tuple)) else [str(cmd)]
        if seq and seq[0].startswith("/command/"):
            seq[0] = seq[0][len("/command/"):]
        svc_calls.append(seq)
        if seq and seq[0] == "s6-svstat":
            return _sp.CompletedProcess(cmd, 0, "up (pid 9090) 5 seconds\n", "")
        return _sp.CompletedProcess(cmd, 0, "", "")

    monkeypatch.setattr("subprocess.run", _fake)

    marked: list[int] = []
    monkeypatch.setattr(
        "gateway.status.write_planned_stop_marker",
        lambda pid: marked.append(pid) or True,
    )

    mgr = S6ServiceManager(scandir=s6_scandir)
    mgr.stop("gateway-coder")

    assert marked == [9090], (
        f"stop() must write the planned-stop marker for the supervised PID; "
        f"marked={marked}"
    )
    # And it must still issue the down command.
    assert any(
        cmd[0] == "s6-svc" and "-d" in cmd for cmd in svc_calls
    ), f"s6-svc -d not invoked; saw: {svc_calls}"


def test_s6_stop_tolerates_marker_write_failure(monkeypatch, s6_scandir):
    """A marker-write failure must not block the stop (best-effort)."""
    import subprocess as _sp

    svc_dir = s6_scandir / "gateway-coder"
    svc_dir.mkdir()

    svc_calls: list[list[str]] = []

    def _fake(cmd, **kw):
        seq = list(cmd) if isinstance(cmd, (list, tuple)) else [str(cmd)]
        if seq and seq[0].startswith("/command/"):
            seq[0] = seq[0][len("/command/"):]
        svc_calls.append(seq)
        if seq and seq[0] == "s6-svstat":
            return _sp.CompletedProcess(cmd, 0, "up (pid 9090) 5 seconds\n", "")
        return _sp.CompletedProcess(cmd, 0, "", "")

    monkeypatch.setattr("subprocess.run", _fake)

    def _boom(pid):
        raise OSError("disk full")

    monkeypatch.setattr("gateway.status.write_planned_stop_marker", _boom)

    mgr = S6ServiceManager(scandir=s6_scandir)
    mgr.stop("gateway-coder")  # must not raise

    assert any(cmd[0] == "s6-svc" and "-d" in cmd for cmd in svc_calls)


def test_s6_log_run_chowns_gateways_parent(s6_scandir, fake_subprocess_run) -> None:
    """The log/run script must chown the logs/gateways/ parent, not just the leaf.

    Regression guard for #45258: `mkdir -p` creates the gateways/ parent
    root-owned on a root-context boot, and a leaf-only chown leaves it that
    way. Every profile registered later then runs its log service as the
    dropped hermes user and s6-log crash-loops on `mkdir: Permission denied`.
    """
    mgr = S6ServiceManager(scandir=s6_scandir)
    mgr.register_profile_gateway("coder")

    log_text = (s6_scandir / "gateway-coder" / "log" / "run").read_text()

    parent_chown = 'chown hermes:hermes "$HERMES_HOME/logs/gateways"'
    assert parent_chown in log_text, (
        "log/run must chown the logs/gateways parent so profiles added "
        f"after a root-context boot can create their leaf dirs. Saw: {log_text!r}"
    )
    # Non-recursive on purpose: sibling profile leaf dirs are each managed
    # by their own log/run; a recursive parent chown would race them.
    assert 'chown -R hermes:hermes "$HERMES_HOME/logs/gateways"' not in log_text

    # Ordering: mkdir creates the parent, then the parent chown repairs its
    # ownership, then the leaf chown — all before s6-log execs.
    mkdir_idx = log_text.index('mkdir -p "$log_dir"')
    parent_idx = log_text.index(parent_chown)
    leaf_idx = log_text.index('chown -R hermes:hermes "$log_dir"')
    exec_idx = log_text.index("s6-log 1 ")
    assert mkdir_idx < parent_idx < leaf_idx < exec_idx

    # The parent path must be a runtime env expansion, never a baked-in
    # absolute path (same contract as the log_dir itself).
    assert '/opt/data/logs/gateways"' not in log_text