Merge pull request #31760 from NousResearch/hermes/hermes-bf5898da

feat(docker)!: s6-overlay container supervision (salvage of #30136)

tests/docker/conftest.py

@@ -0,0 +1,139 @@
+"""Shared fixtures for docker-image integration tests.
+
+Tests in this directory build the image with the current ``Dockerfile``
+and exercise it via ``docker run``. They skip when Docker is unavailable
+(e.g. on developer laptops without a daemon).
+
+Override the image with ``HERMES_TEST_IMAGE`` env var to point at a pre-built
+image (faster local iteration); otherwise the ``built_image`` fixture builds
+the repo's Dockerfile once per session.
+
+Docker tests need longer timeouts than the suite default (30s), so every
+test under this directory is granted a 180s default via
+``pytest.mark.timeout`` applied at collection time.
+"""
+from __future__ import annotations
+
+import os
+import shutil
+import subprocess
+from collections.abc import Iterator
+
+import pytest
+
+IMAGE_TAG = os.environ.get("HERMES_TEST_IMAGE", "hermes-agent-harness:latest")
+
+
+def _docker_available() -> bool:
+    """Return True iff a docker CLI is on PATH and the daemon answers."""
+    if shutil.which("docker") is None:
+        return False
+    try:
+        r = subprocess.run(
+            ["docker", "info"], capture_output=True, timeout=5,
+        )
+        return r.returncode == 0
+    except (subprocess.TimeoutExpired, OSError):
+        return False
+
+
+def pytest_collection_modifyitems(config, items):  # noqa: D401 - pytest hook
+    """Apply docker-suite policy: timeout bump + skip on missing docker."""
+    docker_ok = _docker_available()
+    skip_docker = pytest.mark.skip(
+        reason="Docker not available or daemon not running",
+    )
+    extend_timeout = pytest.mark.timeout(180)
+    for item in items:
+        if "tests/docker/" not in str(item.fspath).replace(os.sep, "/"):
+            continue
+        item.add_marker(extend_timeout)
+        if not docker_ok:
+            item.add_marker(skip_docker)
+
+
+@pytest.fixture(scope="session")
+def built_image() -> str:
+    """Build the image once per test session.
+
+    Override with ``HERMES_TEST_IMAGE`` env var to point at a pre-built
+    image (faster local iteration).
+    """
+    if os.environ.get("HERMES_TEST_IMAGE"):
+        return IMAGE_TAG
+    repo_root = os.path.abspath(
+        os.path.join(os.path.dirname(__file__), "..", ".."),
+    )
+    result = subprocess.run(
+        ["docker", "build", "-t", IMAGE_TAG, repo_root],
+        capture_output=True, text=True, timeout=1200,
+    )
+    assert result.returncode == 0, (
+        f"docker build failed:\n{result.stderr[-2000:]}"
+    )
+    return IMAGE_TAG
+
+
+@pytest.fixture
+def container_name(request) -> Iterator[str]:
+    """Generate a unique container name and ensure cleanup on test exit."""
+    safe = request.node.name.replace("[", "_").replace("]", "_")
+    name = f"hermes-test-{safe}"
+    yield name
+    subprocess.run(
+        ["docker", "rm", "-f", name],
+        capture_output=True, timeout=10,
+    )
+
+
+# ---------------------------------------------------------------------------
+# docker_exec — default to the unprivileged hermes user
+# ---------------------------------------------------------------------------
+#
+# Background: every Hermes runtime path inside the container drops to UID
+# 10000 (the ``hermes`` user) via ``s6-setuidgid hermes``. ``docker exec``
+# without ``-u`` runs as root, which is **not** representative of how
+# production code executes. PR #30136 review caught a real regression
+# this way — ``Path('/proc/1/exe').resolve()`` works as root and silently
+# fails (PermissionError swallowed) for hermes, so a test that ran as root
+# couldn't catch a feature that was inert for the actual runtime user.
+#
+# Tests in this directory MUST exercise the realistic user context. The
+# helpers below run every probe under ``-u hermes`` unless a specific
+# test explicitly opts into ``user="root"`` (rare — e.g. inspecting
+# /proc/1/exe itself, chowning a volume).
+# ---------------------------------------------------------------------------
+
+
+def docker_exec(
+    container: str,
+    *args: str,
+    user: str = "hermes",
+    timeout: int = 30,
+    extra_docker_args: tuple[str, ...] = (),
+) -> subprocess.CompletedProcess[str]:
+    """Run a command inside ``container`` as ``user`` (default: hermes).
+
+    Returns the CompletedProcess with text=True, capture_output=True.
+
+    Pass ``user="root"`` only when the test specifically needs root
+    capabilities (e.g. reading /proc/1/exe, manipulating ownership).
+    Most tests should use the default.
+    """
+    cmd = ["docker", "exec", "-u", user, *extra_docker_args, container, *args]
+    return subprocess.run(
+        cmd, capture_output=True, text=True, timeout=timeout,
+    )
+
+
+def docker_exec_sh(
+    container: str,
+    command: str,
+    *,
+    user: str = "hermes",
+    timeout: int = 30,
+) -> subprocess.CompletedProcess[str]:
+    """Run ``sh -c <command>`` inside the container as ``user``."""
+    return docker_exec(
+        container, "sh", "-c", command, user=user, timeout=timeout,
+    )

tests/docker/test_container_restart.py

@@ -0,0 +1,252 @@
+"""Container-restart survives per-profile gateway registrations.
+
+The s6 dynamic scandir at /run/service/ lives on tmpfs and is wiped
+on every container restart. Phase 4 Task 4.0's container_boot module
++ cont-init.d/02-reconcile-profiles regenerate the service slots from
+$HERMES_HOME/profiles/<name>/gateway_state.json on every boot and
+auto-start only those whose last state was `running`.
+
+These tests stand up a container with a named volume, create profiles
+inside it in various gateway states, restart the container, and
+assert the reconciler did the right thing.
+
+Every ``docker exec`` here runs as the unprivileged ``hermes`` user
+(via :func:`docker_exec` / :func:`docker_exec_sh` in conftest); see
+the conftest module docstring.
+"""
+from __future__ import annotations
+
+import subprocess
+import time
+
+import pytest
+
+from tests.docker.conftest import docker_exec, docker_exec_sh
+
+
+def _docker(*args: str, **kw) -> subprocess.CompletedProcess[str]:
+    return subprocess.run(
+        ["docker", *args],
+        capture_output=True, text=True, timeout=kw.pop("timeout", 60),
+        **kw,
+    )
+
+
+def _exec(container: str, *args: str, timeout: int = 30) -> subprocess.CompletedProcess[str]:
+    return docker_exec(container, *args, timeout=timeout)
+
+
+def _sh(container: str, cmd: str, timeout: int = 30) -> subprocess.CompletedProcess[str]:
+    return docker_exec_sh(container, cmd, timeout=timeout)
+
+
+def _wait_for_path(
+    container: str,
+    path: str,
+    *,
+    kind: str = "f",
+    deadline_s: float = 30.0,
+    interval_s: float = 0.25,
+) -> bool:
+    """Poll `test -<kind> <path>` inside container until success or timeout.
+
+    `kind` is the `test` flag: 'f' for file, 'd' for directory, 'e' for
+    existence. Returns True on success, False on timeout. Strictly
+    better than a fixed `time.sleep()` because:
+
+      * we don't wait the full budget when the path appears early, and
+      * the test fails with a precise "waited N seconds" assertion
+        instead of a confusing one-line failure mid-test when the
+        sleep was too short.
+    """
+    end = time.monotonic() + deadline_s
+    while time.monotonic() < end:
+        r = _sh(container, f"test -{kind} {path}", timeout=5)
+        if r.returncode == 0:
+            return True
+        time.sleep(interval_s)
+    return False
+
+
+def _wait_for_reconcile_log_mention(
+    container: str,
+    profile: str,
+    *,
+    deadline_s: float = 30.0,
+    interval_s: float = 0.25,
+) -> str:
+    """Poll until /opt/data/logs/container-boot.log mentions `profile`.
+
+    Returns the matching log content on success. On timeout, returns
+    the last observed contents so the assertion can render a
+    meaningful diagnostic. The container-boot.log is the explicit
+    signal that the reconciler has finished — much more reliable
+    than a fixed sleep that hopes 8 seconds is enough.
+    """
+    end = time.monotonic() + deadline_s
+    last = ""
+    while time.monotonic() < end:
+        r = _sh(container, "cat /opt/data/logs/container-boot.log", timeout=5)
+        if r.returncode == 0:
+            last = r.stdout
+            if f"profile={profile}" in last:
+                return last
+        time.sleep(interval_s)
+    return last
+
+
+@pytest.fixture
+def restart_container(request, built_image: str):
+    """A long-running container with a named volume so docker restart
+    preserves $HERMES_HOME/profiles/."""
+    safe = request.node.name.replace("[", "_").replace("]", "_")
+    name = f"hermes-restart-{safe}"
+    volume = f"hermes-restart-vol-{safe}"
+    _docker("rm", "-f", name)
+    _docker("volume", "rm", "-f", volume)
+    _docker("volume", "create", volume, timeout=10).check_returncode()
+    r = _docker(
+        "run", "-d", "--name", name,
+        "-v", f"{volume}:/opt/data",
+        built_image, "sleep", "infinity",
+        timeout=30,
+    )
+    r.check_returncode()
+    # Wait for s6 + stage2 + 02-reconcile to publish the boot log so
+    # the test can rely on the default slot being registered before
+    # it starts issuing commands. The reconciler always writes one
+    # 'default' line on every boot (PR #30136 item I1) — that's our
+    # readiness signal.
+    deadline = time.monotonic() + 30.0
+    while time.monotonic() < deadline:
+        r = _docker(
+            "exec", "-u", "hermes", name, "sh", "-c",
+            "cat /opt/data/logs/container-boot.log 2>/dev/null",
+            timeout=5,
+        )
+        if r.returncode == 0 and "profile=default" in r.stdout:
+            break
+        time.sleep(0.25)
+    else:
+        # Defensive: surface a timeout from the fixture itself so the
+        # test failure points at "container never finished cont-init"
+        # rather than mid-test where the symptom would be obscure.
+        raise RuntimeError(
+            f"container {name} did not finish cont-init within 30s"
+        )
+    yield name
+    _docker("rm", "-f", name)
+    _docker("volume", "rm", "-f", volume)
+
+
+def test_running_gateway_survives_container_restart(restart_container: str) -> None:
+    container = restart_container
+
+    # Create the profile + start its gateway. The Phase 4 hooks
+    # register the s6 service slot during create and the dispatch
+    # path brings it up via s6-svc -u.
+    r = _exec(container, "hermes", "profile", "create", "coder")
+    assert r.returncode == 0, f"profile create failed: {r.stderr}"
+
+    r = _exec(container, "hermes", "-p", "coder", "gateway", "start", timeout=60)
+    assert r.returncode == 0, f"gateway start failed: {r.stderr}"
+
+    # Give the service time to actually come up under supervision.
+    deadline = time.monotonic() + 15.0
+    while time.monotonic() < deadline:
+        r = _sh(container, "/command/s6-svstat /run/service/gateway-coder")
+        if r.returncode == 0 and "up " in r.stdout:
+            break
+        time.sleep(0.5)
+    assert "up " in r.stdout, f"gateway never came up pre-restart: {r.stdout!r}"
+
+    # Persist state so the reconciler will treat the slot as 'running'
+    # post-restart. The gateway process itself writes gateway_state.json
+    # via gateway/status.py — but we don't want to wait for or assert
+    # against the live process here; just stamp the file directly to
+    # exercise the reconciler's contract.
+    write_state = (
+        "import json, pathlib; "
+        "p = pathlib.Path('/opt/data/profiles/coder/gateway_state.json'); "
+        "p.write_text(json.dumps({'gateway_state': 'running', 'timestamp': 1}))"
+    )
+    _exec(container, "python3", "-c", write_state, timeout=10).check_returncode()
+
+    # Restart. After this, /run/service/ is empty until cont-init.d
+    # runs the reconciler. We need to wait long enough for the
+    # reconciler to write coder's entry to the boot log AND for
+    # s6-svscan to spin up the service supervise tree from the
+    # restored slot. Polling the boot log gives us the first signal.
+    _docker("restart", container, timeout=60).check_returncode()
+    log = _wait_for_reconcile_log_mention(container, "coder", deadline_s=30.0)
+    assert "profile=coder" in log, (
+        f"reconciler never logged coder after restart: {log!r}"
+    )
+    assert "action=started" in log
+
+    # Service slot exists.
+    assert _wait_for_path(
+        container, "/run/service/gateway-coder", kind="d", deadline_s=10.0,
+    ), "slot not recreated after restart"
+
+    # No `down` marker — we asked for auto-start.
+    r = _sh(container, "test -f /run/service/gateway-coder/down")
+    assert r.returncode != 0, "down marker present despite prior_state=running"
+
+
+def test_stopped_gateway_stays_stopped_after_restart(restart_container: str) -> None:
+    container = restart_container
+
+    _exec(container, "hermes", "profile", "create", "writer").check_returncode()
+
+    # Write 'stopped' directly so we don't have to race against the
+    # gateway's own state writes.
+    write_state = (
+        "import json, pathlib; "
+        "p = pathlib.Path('/opt/data/profiles/writer/gateway_state.json'); "
+        "p.write_text(json.dumps({'gateway_state': 'stopped', 'timestamp': 1}))"
+    )
+    _exec(container, "python3", "-c", write_state, timeout=10).check_returncode()
+
+    _docker("restart", container, timeout=60).check_returncode()
+    log = _wait_for_reconcile_log_mention(container, "writer", deadline_s=30.0)
+    assert "profile=writer" in log
+
+    # Slot exists.
+    assert _wait_for_path(
+        container, "/run/service/gateway-writer", kind="d", deadline_s=10.0,
+    )
+
+    # Down marker present.
+    r = _sh(container, "test -f /run/service/gateway-writer/down")
+    assert r.returncode == 0, "down marker missing despite prior_state=stopped"
+
+
+def test_stale_gateway_pid_cleaned_up_on_restart(restart_container: str) -> None:
+    """A dead container's gateway.pid + processes.json must NOT
+    survive the restart — a numerically-equal live PID in the new
+    container is a different process and would confuse the gateway
+    process-mismatch checks."""
+    container = restart_container
+
+    _exec(container, "hermes", "profile", "create", "ghost").check_returncode()
+
+    # Stamp stale runtime files alongside a 'running' state so the
+    # reconciler walks this profile.
+    stamp = (
+        "import json, pathlib; "
+        "p = pathlib.Path('/opt/data/profiles/ghost'); "
+        "(p / 'gateway_state.json').write_text(json.dumps({'gateway_state': 'stopped', 'timestamp': 1})); "
+        "(p / 'gateway.pid').write_text(json.dumps({'pid': 99999, 'host': 'old'})); "
+        "(p / 'processes.json').write_text('[]')"
+    )
+    _exec(container, "python3", "-c", stamp, timeout=10).check_returncode()
+
+    _docker("restart", container, timeout=60).check_returncode()
+    _wait_for_reconcile_log_mention(container, "ghost", deadline_s=30.0)
+
+    # Stale runtime files swept.
+    r = _sh(container, "test -f /opt/data/profiles/ghost/gateway.pid")
+    assert r.returncode != 0, "stale gateway.pid survived restart"
+    r = _sh(container, "test -f /opt/data/profiles/ghost/processes.json")
+    assert r.returncode != 0, "stale processes.json survived restart"

tests/docker/test_dashboard.py

@@ -0,0 +1,203 @@
+"""Harness: dashboard opt-in via HERMES_DASHBOARD.
+
+Today (tini): dashboard starts once when HERMES_DASHBOARD=1; if it crashes
+it stays dead. After Phase 2 (s6): dashboard starts once; if it crashes
+it is restarted under supervision. The restart-after-crash test lives in
+Phase 2 Task 2.5; this file only locks the opt-in surface (which must
+not change between tini and s6).
+
+Every ``docker exec`` here runs as the unprivileged ``hermes`` user
+(via :func:`docker_exec`/:func:`docker_exec_sh` in conftest), matching
+the realistic runtime context. See the conftest module docstring.
+"""
+from __future__ import annotations
+
+import subprocess
+import time
+
+from tests.docker.conftest import docker_exec, docker_exec_sh
+
+
+def _poll(container: str, probe: str, *, deadline_s: float = 30.0,
+          interval_s: float = 0.5) -> tuple[bool, str]:
+    """Repeatedly run ``probe`` inside the container until it exits 0 or
+    ``deadline_s`` elapses. Returns (success, last stdout)."""
+    end = time.monotonic() + deadline_s
+    last = ""
+    while time.monotonic() < end:
+        r = docker_exec_sh(container, probe, timeout=10)
+        last = r.stdout
+        if r.returncode == 0:
+            return True, last
+        time.sleep(interval_s)
+    return False, last
+
+
+def test_dashboard_not_running_by_default(
+    built_image: str, container_name: str,
+) -> None:
+    """Without HERMES_DASHBOARD, no dashboard process should be running."""
+    subprocess.run(
+        ["docker", "run", "-d", "--name", container_name, built_image,
+         "sleep", "60"],
+        check=True, capture_output=True, timeout=30,
+    )
+    # Give the entrypoint enough time to finish bootstrap; if a dashboard
+    # were going to start it'd be visible by now.
+    time.sleep(5)
+    r = docker_exec(container_name, "pgrep", "-f", "hermes dashboard")
+    # pgrep exits non-zero when no match found
+    assert r.returncode != 0, (
+        "Dashboard should not be running without HERMES_DASHBOARD"
+    )
+
+
+def test_dashboard_slot_reports_down_when_disabled(
+    built_image: str, container_name: str,
+) -> None:
+    """Without HERMES_DASHBOARD, s6-svstat should report the dashboard
+    slot as DOWN (not up-with-sleep-infinity, which would
+    false-positive `hermes doctor` and any other health check).
+
+    Locks the PR #30136 review item I3 fix: cont-init.d/03-dashboard-toggle
+    writes a `down` marker file in the live service-dir when
+    HERMES_DASHBOARD is unset, so the slot reflects reality.
+    """
+    subprocess.run(
+        ["docker", "run", "-d", "--name", container_name, built_image,
+         "sleep", "60"],
+        check=True, capture_output=True, timeout=30,
+    )
+    time.sleep(5)
+    # /command/ isn't on PATH for docker-exec sessions, so call by
+    # absolute path.
+    r = docker_exec(
+        container_name, "/command/s6-svstat", "/run/service/dashboard",
+    )
+    assert r.returncode == 0, f"s6-svstat failed: {r.stderr!r} / {r.stdout!r}"
+    assert "down" in r.stdout, (
+        f"Dashboard slot should be 'down' without HERMES_DASHBOARD; "
+        f"svstat reports: {r.stdout!r}"
+    )
+
+
+def test_dashboard_slot_reports_up_when_enabled(
+    built_image: str, container_name: str,
+) -> None:
+    """Symmetry: with HERMES_DASHBOARD=1, s6-svstat reports the slot as up."""
+    subprocess.run(
+        ["docker", "run", "-d", "--name", container_name,
+         "-e", "HERMES_DASHBOARD=1", built_image, "sleep", "120"],
+        check=True, capture_output=True, timeout=30,
+    )
+    # uvicorn takes a moment to bind; poll svstat.
+    deadline = time.monotonic() + 30.0
+    last = ""
+    while time.monotonic() < deadline:
+        r = docker_exec(
+            container_name, "/command/s6-svstat", "/run/service/dashboard",
+        )
+        last = r.stdout
+        if r.returncode == 0 and "up " in r.stdout:
+            return  # success
+        time.sleep(0.5)
+    raise AssertionError(
+        f"Dashboard slot never reached up state; last svstat: {last!r}"
+    )
+
+
+def test_dashboard_opt_in_starts(
+    built_image: str, container_name: str,
+) -> None:
+    """With HERMES_DASHBOARD=1, a dashboard process should be visible."""
+    subprocess.run(
+        ["docker", "run", "-d", "--name", container_name,
+         "-e", "HERMES_DASHBOARD=1", built_image, "sleep", "120"],
+        check=True, capture_output=True, timeout=30,
+    )
+    # Poll for the dashboard subprocess to appear — the entrypoint
+    # backgrounds it and bootstrap (skills sync etc.) can take a few
+    # seconds before the python process actually launches.
+    ok, _ = _poll(
+        container_name, "pgrep -f 'hermes dashboard'", deadline_s=30.0,
+    )
+    assert ok, "Dashboard should be running with HERMES_DASHBOARD=1"
+
+
+def test_dashboard_port_override(
+    built_image: str, container_name: str,
+) -> None:
+    """HERMES_DASHBOARD_PORT changes the dashboard's listen port."""
+    subprocess.run(
+        ["docker", "run", "-d", "--name", container_name,
+         "-e", "HERMES_DASHBOARD=1", "-e", "HERMES_DASHBOARD_PORT=9120",
+         built_image, "sleep", "120"],
+        check=True, capture_output=True, timeout=30,
+    )
+    # The dashboard process appearing in pgrep doesn't mean it's bound
+    # to the port yet — uvicorn takes another second or two to come up.
+    # The image doesn't ship ss/netstat, so probe /proc/net/tcp directly:
+    # port 9120 = 0x23A0, state 0A = LISTEN.
+    ok, stdout = _poll(
+        container_name,
+        "grep -E ' 0+:23A0 .* 0A ' /proc/net/tcp /proc/net/tcp6 "
+        "2>/dev/null",
+        deadline_s=60.0,
+    )
+    assert ok, f"Dashboard not listening on port 9120: stdout={stdout!r}"
+
+
+def test_dashboard_restarts_after_crash(
+    built_image: str, container_name: str,
+) -> None:
+    """Phase 2 invariant: under s6 supervision, killing the dashboard
+    process should be recovered automatically.
+
+    Pre-s6 (tini) behavior was "stays dead" — the test wouldn't have
+    passed against that image. After the s6-overlay migration the
+    dashboard runs as a longrun s6-rc service and s6-supervise restarts
+    it after a ~1s backoff (the default).
+    """
+    subprocess.run(
+        ["docker", "run", "-d", "--name", container_name,
+         "-e", "HERMES_DASHBOARD=1", built_image, "sleep", "120"],
+        check=True, capture_output=True, timeout=30,
+    )
+    # Wait for the first dashboard to come up.
+    ok, _ = _poll(
+        container_name, "pgrep -f 'hermes dashboard'", deadline_s=30.0,
+    )
+    assert ok, "Dashboard never started initially"
+
+    # Grab the initial PID. s6 may briefly transition through restart
+    # state between our poll-success and the follow-up pgrep, so retry
+    # a couple of times before giving up.
+    first_pid: str | None = None
+    for _attempt in range(10):
+        first_pid_result = docker_exec(
+            container_name, "pgrep", "-f", "hermes dashboard",
+        )
+        first_pids = first_pid_result.stdout.strip().split()
+        if first_pids:
+            first_pid = first_pids[0]
+            break
+        time.sleep(0.5)
+    assert first_pid is not None, "Could not capture initial dashboard PID"
+
+    # Kill the dashboard. The dashboard process runs as hermes, so the
+    # hermes user can kill it (same UID).
+    docker_exec(container_name, "kill", "-9", first_pid)
+
+    # s6 backs off ~1s before restart; allow up to 15s for the new
+    # process to appear with a different PID.
+    deadline = time.monotonic() + 15.0
+    while time.monotonic() < deadline:
+        r = docker_exec(container_name, "pgrep", "-f", "hermes dashboard")
+        pids = r.stdout.strip().split() if r.returncode == 0 else []
+        if pids and pids[0] != first_pid:
+            return  # success
+        time.sleep(0.5)
+
+    raise AssertionError(
+        f"Dashboard not restarted after kill (first_pid={first_pid})"
+    )

tests/docker/test_main_invocation.py

@@ -0,0 +1,79 @@
+"""Harness: docker run <image> [cmd...] invocation patterns.
+
+These tests MUST pass on the current tini-based image AND continue to
+pass after the Phase 2 s6 migration. Any behavior drift is a regression.
+
+The harness expects ``built_image`` and ``container_name`` fixtures from
+``tests/docker/conftest.py``. When Docker isn't available every test
+here is skipped at collection time.
+"""
+from __future__ import annotations
+
+import subprocess
+
+
+def test_no_args_starts_hermes(built_image: str) -> None:
+    """``docker run <image>`` should start hermes cleanly.
+
+    We invoke ``--version`` so the call exits without needing a configured
+    model. Exit code may be 0 (printed version) or 1 (config bootstrapping
+    failure on a fresh volume), but never a stack trace.
+    """
+    r = subprocess.run(
+        ["docker", "run", "--rm", built_image, "--version"],
+        capture_output=True, text=True, timeout=60,
+    )
+    assert r.returncode in (0, 1), (
+        f"Unexpected exit {r.returncode}: stderr={r.stderr!r}"
+    )
+    assert "Traceback" not in r.stderr
+
+
+def test_chat_subcommand_passthrough(built_image: str) -> None:
+    """``docker run <image> chat --help`` should exec ``hermes chat --help``.
+
+    Uses ``--help`` so the call doesn't need an upstream model configured.
+    """
+    r = subprocess.run(
+        ["docker", "run", "--rm", built_image, "chat", "--help"],
+        capture_output=True, text=True, timeout=60,
+    )
+    assert r.returncode == 0
+    combined = (r.stdout + r.stderr).lower()
+    assert "chat" in combined or "usage" in combined
+
+
+def test_bare_executable_passthrough(built_image: str) -> None:
+    """``docker run <image> sleep 1`` should exec ``sleep`` directly.
+
+    The entrypoint detects that ``sleep`` is on PATH and routes around the
+    hermes wrapper. Useful for long-lived sandbox mode and for testing.
+    """
+    r = subprocess.run(
+        ["docker", "run", "--rm", built_image, "sleep", "1"],
+        capture_output=True, text=True, timeout=30,
+    )
+    assert r.returncode == 0
+
+
+def test_bash_pattern(built_image: str) -> None:
+    """``docker run <image> bash -c 'echo ok'`` should exec bash directly."""
+    r = subprocess.run(
+        ["docker", "run", "--rm", built_image, "bash", "-c", "echo ok"],
+        capture_output=True, text=True, timeout=30,
+    )
+    assert r.returncode == 0
+    assert "ok" in r.stdout
+
+
+def test_container_exit_code_matches_inner_exit(built_image: str) -> None:
+    """The container exit code must match the inner process's exit code.
+
+    Critical for CI: ``docker run <image> hermes batch ...`` returns a
+    non-zero status when batch fails. Phase 2 (s6) must preserve this.
+    """
+    r = subprocess.run(
+        ["docker", "run", "--rm", built_image, "sh", "-c", "exit 42"],
+        capture_output=True, text=True, timeout=30,
+    )
+    assert r.returncode == 42

tests/docker/test_profile_gateway.py

@@ -0,0 +1,138 @@
+"""Harness: per-profile gateway start/stop inside the container.
+
+Phase 4 wires `hermes -p <profile> gateway start/stop` through the s6
+ServiceManager dispatch path inside the container — so the lifecycle
+commands now bring up an s6-supervised gateway rather than refusing
+with the pre-Phase-4 informational message.
+
+These tests were marked ``xfail(strict=True)`` through Phase 0–3 and
+flip to plain ``test_…`` once Phase 4 lands (now).
+
+NB: The harness profile has no model/auth configured. Depending on
+how the gateway run script handles missing config, the supervised
+process may either spin up successfully (and svstat reports ``up``)
+or exit fast and get throttled by s6 (and svstat reports ``down …,
+want up``). Both states are valid "user asked for gateway up" results
+— what we assert is the *want* intent the lifecycle command set, NOT
+the supervised process's health. ``s6-svc -u`` records ``want up`` in
+the supervise/status file regardless of the run-script outcome.
+
+Every ``docker exec`` here runs as the unprivileged ``hermes`` user
+(via :func:`docker_exec_sh` in conftest); see the conftest module
+docstring.
+"""
+from __future__ import annotations
+
+import subprocess
+import time
+
+from tests.docker.conftest import docker_exec_sh
+
+PROFILE = "test-harness-profile"
+
+
+def _sh(
+    container: str, command: str, timeout: int = 30,
+) -> subprocess.CompletedProcess[str]:
+    return docker_exec_sh(container, command, timeout=timeout)
+
+
+def _svstat(container: str) -> str:
+    """Returns the raw s6-svstat output for the test profile's slot.
+    /command/s6-svstat is called by absolute path because /command/
+    isn't on PATH for docker-exec sessions."""
+    r = _sh(container, f"/command/s6-svstat /run/service/gateway-{PROFILE}")
+    return r.stdout if r.returncode == 0 else ""
+
+
+def _svstat_wants_up(container: str) -> bool:
+    """Read the slot's want-state from s6-svstat output.
+
+    s6-svstat formats the output to elide redundancies — when the
+    service is currently up AND s6 wants it up, the literal token
+    ``want up`` doesn't appear (it's implicit from the leading ``up``).
+    When the service is down but s6 wants it back up, ``, want up``
+    appears explicitly. So a comprehensive "is the want-intent set to
+    up" check has to accept both spellings.
+    """
+    state = _svstat(container)
+    if not state:
+        return False
+    head = state.split()[0] if state.split() else ""
+    if head == "up":
+        # Currently up implies wanted-up unless ``want down`` is set.
+        return "want down" not in state
+    # Currently down — ``want up`` only shows up when explicitly set.
+    return "want up" in state
+
+
+def test_profile_create_then_gateway_start(
+    built_image: str, container_name: str,
+) -> None:
+    subprocess.run(
+        ["docker", "run", "-d", "--name", container_name, built_image,
+         "sleep", "120"],
+        check=True, capture_output=True, timeout=30,
+    )
+    time.sleep(3)
+
+    r = _sh(container_name, f"hermes profile create {PROFILE}")
+    assert r.returncode == 0, f"profile create failed: {r.stderr}"
+
+    # Profile create's s6-register hook should have produced a service slot.
+    r = _sh(container_name, f"test -d /run/service/gateway-{PROFILE}")
+    assert r.returncode == 0, "s6 service slot not created on profile create"
+
+    r = _sh(container_name, f"hermes -p {PROFILE} gateway start", timeout=60)
+    assert r.returncode == 0, (
+        f"gateway start failed: stderr={r.stderr!r} stdout={r.stdout!r}"
+    )
+
+    # After start, s6's intent is "up" — even if the supervised gateway
+    # process spin-fails (no model/auth in the test profile), the
+    # supervision-state contract holds. See ``_svstat_wants_up`` for
+    # why we accept both ``up …`` (currently up) and ``down …, want
+    # up`` (down but s6 wants up).
+    time.sleep(2)
+    assert _svstat_wants_up(container_name), (
+        f"slot want-state is not up after gateway start: "
+        f"{_svstat(container_name)!r}"
+    )
+
+    r = _sh(container_name, f"hermes -p {PROFILE} gateway stop", timeout=30)
+    assert r.returncode == 0
+
+    time.sleep(2)
+    assert not _svstat_wants_up(container_name), (
+        f"slot want-state still up after gateway stop: "
+        f"{_svstat(container_name)!r}"
+    )
+
+
+def test_profile_delete_stops_gateway(
+    built_image: str, container_name: str,
+) -> None:
+    """Deleting a profile should stop its gateway and remove the s6
+    service slot."""
+    subprocess.run(
+        ["docker", "run", "-d", "--name", container_name, built_image,
+         "sleep", "120"],
+        check=True, capture_output=True, timeout=30,
+    )
+    time.sleep(3)
+
+    _sh(container_name, f"hermes profile create {PROFILE}")
+    _sh(container_name, f"hermes -p {PROFILE} gateway start", timeout=60)
+    time.sleep(3)
+
+    r = _sh(
+        container_name,
+        f"hermes profile delete {PROFILE} --yes",
+        timeout=30,
+    )
+    assert r.returncode == 0, f"profile delete failed: {r.stderr}"
+
+    time.sleep(2)
+    # Service slot should be gone.
+    r = _sh(container_name, f"test -d /run/service/gateway-{PROFILE}")
+    assert r.returncode != 0, "s6 service slot still present after profile delete"

tests/docker/test_s6_profile_gateway_integration.py

@@ -0,0 +1,129 @@
+"""Harness: in-container integration tests for S6ServiceManager.
+
+The unit tests in tests/hermes_cli/test_service_manager.py exercise the
+class against a tmp-path scandir with a stubbed ``subprocess.run``.
+These tests run the real class inside a real container against the
+real s6-svc / s6-svscanctl binaries, validating end-to-end.
+
+Phase 3 only registers the service slot — it doesn't depend on the
+gateway actually starting (the binary will refuse to start without a
+valid profile config). The full register → start → supervised-restart
+→ unregister cycle is covered by Phase 4 once profile create/delete
+hooks land.
+
+Every ``docker exec`` here runs as the unprivileged ``hermes`` user
+(via :func:`docker_exec` in conftest); see the conftest module
+docstring. ``/run/service`` is chowned hermes-writable by the
+``02-reconcile-profiles`` cont-init.d script, so register/unregister
+operations work correctly under UID 10000.
+"""
+from __future__ import annotations
+
+import subprocess
+import time
+
+from tests.docker.conftest import docker_exec
+
+
+_REGISTER_SCRIPT = """
+import sys
+sys.path.insert(0, "/opt/hermes")
+from hermes_cli.service_manager import S6ServiceManager
+S6ServiceManager().register_profile_gateway("phase3test")
+# Don't worry about whether the gateway actually starts — we only care
+# that the supervision slot was created. The gateway run script will
+# likely error out (no profile config exists) but that's expected.
+print("REGISTERED")
+"""
+
+_UNREGISTER_SCRIPT = """
+import sys
+sys.path.insert(0, "/opt/hermes")
+from hermes_cli.service_manager import S6ServiceManager
+S6ServiceManager().unregister_profile_gateway("phase3test")
+print("UNREGISTERED")
+"""
+
+
+def _exec(container: str, *args: str, timeout: int = 30) -> subprocess.CompletedProcess:
+    return docker_exec(container, *args, timeout=timeout)
+
+
+def test_s6_register_creates_service_dir_in_live_container(
+    built_image: str, container_name: str,
+) -> None:
+    """S6ServiceManager.register_profile_gateway must create
+    ``/run/service/gateway-<profile>/`` and trigger s6-svscan rescan
+    against the real s6 supervision tree."""
+    subprocess.run(
+        ["docker", "run", "-d", "--name", container_name, built_image,
+         "sleep", "120"],
+        check=True, capture_output=True, timeout=30,
+    )
+    # Give the supervision tree a moment to come up.
+    time.sleep(3)
+
+    r = _exec(container_name, "python3", "-c", _REGISTER_SCRIPT, timeout=30)
+    assert "REGISTERED" in r.stdout, (
+        f"register failed: stderr={r.stderr!r} stdout={r.stdout!r}"
+    )
+
+    # Service directory exists with the expected structure.
+    r = _exec(container_name, "test", "-d", "/run/service/gateway-phase3test")
+    assert r.returncode == 0, "service directory not created"
+
+    r = _exec(container_name, "test", "-f", "/run/service/gateway-phase3test/run")
+    assert r.returncode == 0, "run script not created"
+
+    r = _exec(container_name, "test", "-f",
+              "/run/service/gateway-phase3test/log/run")
+    assert r.returncode == 0, "log/run script not created"
+
+    # s6-svscan picked it up — s6-svstat works against the dir.
+    # `docker exec` doesn't put /command/ on PATH (only the supervision
+    # tree does), so call s6-svstat by absolute path.
+    r = _exec(container_name, "/command/s6-svstat",
+              "/run/service/gateway-phase3test")
+    assert r.returncode == 0, f"s6-svstat failed: {r.stderr or r.stdout}"
+
+    # list_profile_gateways picks it up.
+    r = _exec(container_name, "python3", "-c", (
+        "from hermes_cli.service_manager import S6ServiceManager;"
+        "print(S6ServiceManager().list_profile_gateways())"
+    ))
+    assert "phase3test" in r.stdout, f"list output: {r.stdout!r}"
+
+
+def test_s6_unregister_removes_service_dir_in_live_container(
+    built_image: str, container_name: str,
+) -> None:
+    """unregister_profile_gateway must stop the service, remove the
+    directory, and trigger s6-svscan rescan so the supervise process
+    is dropped."""
+    subprocess.run(
+        ["docker", "run", "-d", "--name", container_name, built_image,
+         "sleep", "120"],
+        check=True, capture_output=True, timeout=30,
+    )
+    time.sleep(3)
+
+    # First register so we have something to unregister.
+    r = _exec(container_name, "python3", "-c", _REGISTER_SCRIPT, timeout=30)
+    assert "REGISTERED" in r.stdout
+
+    # Then unregister.
+    r = _exec(container_name, "python3", "-c", _UNREGISTER_SCRIPT, timeout=30)
+    assert "UNREGISTERED" in r.stdout, (
+        f"unregister failed: stderr={r.stderr!r} stdout={r.stdout!r}"
+    )
+
+    # Directory is gone.
+    r = _exec(container_name, "test", "-d", "/run/service/gateway-phase3test")
+    assert r.returncode != 0, "service directory still exists after unregister"
+
+    # list_profile_gateways no longer includes it.
+    r = _exec(container_name, "python3", "-c", (
+        "from hermes_cli.service_manager import S6ServiceManager;"
+        "print(S6ServiceManager().list_profile_gateways())"
+    ))
+    assert "phase3test" not in r.stdout

tests/docker/test_tui_passthrough.py

@@ -0,0 +1,51 @@
+"""Harness: interactive TUI TTY passthrough.
+
+Uses ``script -qc`` on the host to allocate a PTY for the docker client,
+which then allocates a container-side PTY via ``-t``. The probe inside
+the container is ``tput cols``, which returns a real column count when
+stdout is a TTY and either prints ``80`` (the terminfo fallback) or
+nothing when it is not.
+
+These tests MUST pass on the current tini-based image AND continue to
+pass after the Phase 2 s6 migration. Any drift is a regression.
+"""
+from __future__ import annotations
+
+import shlex
+import shutil
+import subprocess
+
+import pytest
+
+pytestmark = pytest.mark.skipif(
+    shutil.which("script") is None,
+    reason="`script` command not available on this host",
+)
+
+
+def test_tty_passthrough_to_container(built_image: str) -> None:
+    """``docker run -t`` must deliver a real TTY to the container process."""
+    probe = "if [ -t 1 ]; then tput cols; else echo NO_TTY; fi"
+    cmd = (
+        f"docker run --rm -t -e COLUMNS=123 {built_image} "
+        f"sh -c {shlex.quote(probe)}"
+    )
+    r = subprocess.run(
+        ["script", "-qc", cmd, "/dev/null"],
+        capture_output=True, text=True, timeout=120,
+    )
+    output = r.stdout.strip()
+    assert "NO_TTY" not in output, f"TTY passthrough failed: {output!r}"
+    numeric_lines = [s for s in output.split() if s.strip().isdigit()]
+    assert numeric_lines, f"No numeric width in output: {output!r}"
+    assert int(numeric_lines[0]) > 0
+
+
+def test_tui_flag_recognized(built_image: str) -> None:
+    """``docker run -it <image> --help`` should run without crashing."""
+    cmd = f"docker run --rm -t {built_image} --help"
+    r = subprocess.run(
+        ["script", "-qc", cmd, "/dev/null"],
+        capture_output=True, text=True, timeout=60,
+    )
+    assert r.returncode == 0

tests/docker/test_zombie_reaping.py

@@ -0,0 +1,45 @@
+"""Harness: PID 1 must reap orphaned zombie processes.
+
+tini (current PID 1) reaps zombies via its built-in subreaper behavior.
+s6-overlay's ``/init`` (Phase 2 PID 1) does the same. This invariant is
+required for long-running containers spawning subprocesses (subagents,
+dashboard, dynamic gateways) — otherwise the process table fills with
+defunct entries and eventually exhausts the kernel PID space.
+
+Every ``docker exec`` here runs as the unprivileged ``hermes`` user
+(via :func:`docker_exec_sh` in conftest); see the conftest module
+docstring.
+"""
+from __future__ import annotations
+
+import subprocess
+import time
+
+from tests.docker.conftest import docker_exec, docker_exec_sh
+
+
+def test_orphan_zombies_reaped(
+    built_image: str, container_name: str,
+) -> None:
+    """Spawn an orphan child that exits immediately. PID 1 must reap it."""
+    subprocess.run(
+        ["docker", "run", "-d", "--name", container_name, built_image,
+         "sleep", "60"],
+        check=True, capture_output=True, timeout=30,
+    )
+    time.sleep(2)
+
+    # `( ( sleep 0.1 & ) & ); sleep 1` creates a grandchild detached from
+    # the original docker exec session — it becomes an orphan reparented
+    # to PID 1 in the container. When it exits, PID 1 must reap it.
+    docker_exec_sh(
+        container_name, "( ( sleep 0.1 & ) & ); sleep 1", timeout=10,
+    )
+    time.sleep(1)
+
+    r = docker_exec(container_name, "ps", "axo", "stat,pid,comm")
+    zombies = [
+        line for line in r.stdout.split("\n")
+        if line.strip().startswith("Z")
+    ]
+    assert not zombies, f"Zombies not reaped by PID 1: {zombies}"

tests/hermes_cli/test_container_boot.py

@@ -0,0 +1,578 @@
+"""Tests for hermes_cli.container_boot — the cont-init.d-time
+reconciliation that recreates per-profile gateway s6 service slots
+from the persistent profiles directory.
+
+These tests run against a fake $HERMES_HOME under tmp_path; no real
+s6 supervision tree is required. The in-container integration test
+covering end-to-end "docker restart" survival lives in
+tests/docker/test_container_restart.py.
+"""
+from __future__ import annotations
+
+import json
+from pathlib import Path
+
+import pytest
+
+from hermes_cli.container_boot import (
+    ReconcileAction,
+    reconcile_profile_gateways,
+)
+
+
+# ---------------------------------------------------------------------------
+# Fixtures + helpers
+# ---------------------------------------------------------------------------
+
+
+def _make_profile(
+    hermes_home: Path,
+    name: str,
+    *,
+    state: str | None,
+    with_pid: bool = False,
+    config: bool = True,
+) -> Path:
+    """Create a fake profile directory under hermes_home/profiles/<name>/."""
+    p = hermes_home / "profiles" / name
+    p.mkdir(parents=True)
+    if config:
+        # SOUL.md is what the reconciler keys on — it's always seeded by
+        # `hermes profile create`. See container_boot._render_run_script.
+        (p / "SOUL.md").write_text("# fake profile\n")
+    if state is not None:
+        (p / "gateway_state.json").write_text(json.dumps({
+            "gateway_state": state, "timestamp": 1234567890,
+        }))
+    if with_pid:
+        (p / "gateway.pid").write_text(json.dumps(
+            {"pid": 99999, "host": "old-container"},
+        ))
+        (p / "processes.json").write_text("[]")
+    return p
+
+
+def _seed_default_root(
+    hermes_home: Path,
+    *,
+    state: str | None = None,
+    with_pid: bool = False,
+) -> None:
+    """Populate gateway_state.json / stale runtime files at the
+    HERMES_HOME root (the implicit default profile)."""
+    if state is not None:
+        (hermes_home / "gateway_state.json").write_text(json.dumps({
+            "gateway_state": state, "timestamp": 1234567890,
+        }))
+    if with_pid:
+        (hermes_home / "gateway.pid").write_text(json.dumps(
+            {"pid": 99999, "host": "old-container"},
+        ))
+        (hermes_home / "processes.json").write_text("[]")
+
+
+def _named_actions(actions: list[ReconcileAction]) -> list[ReconcileAction]:
+    """Drop the always-present default-profile action so tests that
+    only care about named profiles can assert against a clean list."""
+    return [a for a in actions if a.profile != "default"]
+
+
+# ---------------------------------------------------------------------------
+# Tests
+# ---------------------------------------------------------------------------
+
+
+def test_running_profile_is_registered_and_autostarted(tmp_path: Path) -> None:
+    scandir = tmp_path / "run-service"; scandir.mkdir()
+    _make_profile(tmp_path, "coder", state="running")
+
+    actions = reconcile_profile_gateways(
+        hermes_home=tmp_path, scandir=scandir, dry_run=False,
+    )
+
+    assert _named_actions(actions) == [ReconcileAction(
+        profile="coder", prior_state="running", action="started",
+    )]
+    svc = scandir / "gateway-coder"
+    assert (svc / "run").exists()
+    assert (svc / "run").stat().st_mode & 0o111  # executable
+    assert (svc / "type").read_text().strip() == "longrun"
+    # Auto-start means no down-marker.
+    assert not (svc / "down").exists()
+
+
+def test_stopped_profile_is_registered_but_not_started(tmp_path: Path) -> None:
+    scandir = tmp_path / "run-service"; scandir.mkdir()
+    _make_profile(tmp_path, "writer", state="stopped")
+
+    actions = reconcile_profile_gateways(
+        hermes_home=tmp_path, scandir=scandir, dry_run=False,
+    )
+
+    assert _named_actions(actions) == [ReconcileAction(
+        profile="writer", prior_state="stopped", action="registered",
+    )]
+    # down marker tells s6-svscan to NOT start the service.
+    assert (scandir / "gateway-writer" / "down").exists()
+
+
+def test_startup_failed_does_not_autostart(tmp_path: Path) -> None:
+    """Avoid crash-loop on restart when the gateway was failing to boot."""
+    scandir = tmp_path / "run-service"; scandir.mkdir()
+    _make_profile(tmp_path, "broken", state="startup_failed")
+
+    actions = reconcile_profile_gateways(
+        hermes_home=tmp_path, scandir=scandir, dry_run=False,
+    )
+
+    named = _named_actions(actions)
+    assert named[0].action == "registered"
+    assert (scandir / "gateway-broken" / "down").exists()
+
+
+def test_starting_state_does_not_autostart(tmp_path: Path) -> None:
+    """`starting` means the gateway died mid-boot last time; treat as
+    failed, not as a candidate for auto-restart."""
+    scandir = tmp_path / "run-service"; scandir.mkdir()
+    _make_profile(tmp_path, "unlucky", state="starting")
+
+    actions = reconcile_profile_gateways(
+        hermes_home=tmp_path, scandir=scandir, dry_run=False,
+    )
+
+    named = _named_actions(actions)
+    assert named[0].action == "registered"
+
+
+def test_stale_runtime_files_are_removed(tmp_path: Path) -> None:
+    scandir = tmp_path / "run-service"; scandir.mkdir()
+    profile = _make_profile(tmp_path, "coder", state="running", with_pid=True)
+    assert (profile / "gateway.pid").exists()
+    assert (profile / "processes.json").exists()
+
+    reconcile_profile_gateways(
+        hermes_home=tmp_path, scandir=scandir, dry_run=False,
+    )
+
+    assert not (profile / "gateway.pid").exists()
+    assert not (profile / "processes.json").exists()
+
+
+def test_profile_without_state_file_is_registered_but_not_started(
+    tmp_path: Path,
+) -> None:
+    """A freshly-created profile that's never been started: register
+    its slot but don't auto-start."""
+    scandir = tmp_path / "run-service"; scandir.mkdir()
+    _make_profile(tmp_path, "fresh", state=None)
+
+    actions = reconcile_profile_gateways(
+        hermes_home=tmp_path, scandir=scandir, dry_run=False,
+    )
+
+    assert _named_actions(actions) == [ReconcileAction(
+        profile="fresh", prior_state=None, action="registered",
+    )]
+    assert (scandir / "gateway-fresh" / "down").exists()
+
+
+def test_directory_without_marker_file_is_skipped(tmp_path: Path) -> None:
+    """A stray dir under profiles/ that isn't actually a profile (no
+    SOUL.md — the marker the reconciler keys on) should be skipped."""
+    scandir = tmp_path / "run-service"; scandir.mkdir()
+    # Create a profile dir but without SOUL.md
+    (tmp_path / "profiles" / "stray").mkdir(parents=True)
+
+    actions = reconcile_profile_gateways(
+        hermes_home=tmp_path, scandir=scandir, dry_run=False,
+    )
+
+    assert _named_actions(actions) == []
+    assert not (scandir / "gateway-stray").exists()
+
+
+def test_corrupt_state_file_treated_as_no_prior_state(tmp_path: Path) -> None:
+    """If gateway_state.json is malformed JSON, don't blow up the whole
+    reconciliation — register the slot in the down state."""
+    scandir = tmp_path / "run-service"; scandir.mkdir()
+    profile = _make_profile(tmp_path, "junk", state="running")
+    (profile / "gateway_state.json").write_text("{ not valid json")
+
+    actions = reconcile_profile_gateways(
+        hermes_home=tmp_path, scandir=scandir, dry_run=False,
+    )
+
+    named = _named_actions(actions)
+    assert named[0].action == "registered"  # not "started"
+    assert (scandir / "gateway-junk" / "down").exists()
+
+
+def test_reconcile_log_is_written(tmp_path: Path) -> None:
+    scandir = tmp_path / "run-service"; scandir.mkdir()
+    _make_profile(tmp_path, "a", state="running")
+    _make_profile(tmp_path, "b", state="stopped")
+
+    reconcile_profile_gateways(
+        hermes_home=tmp_path, scandir=scandir, dry_run=False,
+    )
+
+    log = (tmp_path / "logs" / "container-boot.log").read_text()
+    assert "profile=a" in log
+    assert "action=started" in log
+    assert "profile=b" in log
+    assert "action=registered" in log
+
+
+def test_reconcile_log_rotates_when_size_exceeded(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """When container-boot.log exceeds _LOG_ROTATE_BYTES, the existing
+    file is rotated to .1 before the new entries are appended."""
+    from hermes_cli import container_boot
+
+    # Tighten the threshold so we don't have to write 256 KiB.
+    monkeypatch.setattr(container_boot, "_LOG_ROTATE_BYTES", 200)
+
+    log_path = tmp_path / "logs" / "container-boot.log"
+    log_path.parent.mkdir()
+    log_path.write_text("X" * 300)  # already over the threshold
+
+    scandir = tmp_path / "run-service"; scandir.mkdir()
+    _make_profile(tmp_path, "coder", state="running")
+
+    reconcile_profile_gateways(
+        hermes_home=tmp_path, scandir=scandir, dry_run=False,
+    )
+
+    rotated = tmp_path / "logs" / "container-boot.log.1"
+    assert rotated.exists(), "expected previous log to be rotated to .1"
+    assert rotated.read_text().startswith("X" * 300)
+    # The new entries land in a fresh container-boot.log (no leftover Xs).
+    new_contents = log_path.read_text()
+    assert "X" not in new_contents
+    assert "profile=coder" in new_contents
+
+
+def test_reconcile_log_does_not_rotate_below_threshold(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """A small existing log is appended to in place; no .1 is created."""
+    from hermes_cli import container_boot
+    monkeypatch.setattr(container_boot, "_LOG_ROTATE_BYTES", 10_000_000)
+
+    log_path = tmp_path / "logs" / "container-boot.log"
+    log_path.parent.mkdir()
+    log_path.write_text("previous entry\n")
+
+    scandir = tmp_path / "run-service"; scandir.mkdir()
+    _make_profile(tmp_path, "coder", state="running")
+
+    reconcile_profile_gateways(
+        hermes_home=tmp_path, scandir=scandir, dry_run=False,
+    )
+
+    assert not (tmp_path / "logs" / "container-boot.log.1").exists()
+    contents = log_path.read_text()
+    assert contents.startswith("previous entry\n")
+    assert "profile=coder" in contents
+
+
+def test_reconcile_log_rotation_overwrites_existing_dot1(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """Rotating again replaces the prior .1 — we keep at most one
+    rotated file (soft cap of ~2 × threshold)."""
+    from hermes_cli import container_boot
+    monkeypatch.setattr(container_boot, "_LOG_ROTATE_BYTES", 200)
+
+    log_dir = tmp_path / "logs"; log_dir.mkdir()
+    (log_dir / "container-boot.log.1").write_text("OLD ROTATION")
+    (log_dir / "container-boot.log").write_text("Y" * 300)
+
+    scandir = tmp_path / "run-service"; scandir.mkdir()
+    _make_profile(tmp_path, "coder", state="running")
+
+    reconcile_profile_gateways(
+        hermes_home=tmp_path, scandir=scandir, dry_run=False,
+    )
+
+    # .1 now contains the previous .log (Ys), not OLD ROTATION.
+    rotated = (log_dir / "container-boot.log.1").read_text()
+    assert "OLD ROTATION" not in rotated
+    assert rotated.startswith("Y" * 300)
+
+
+def test_dry_run_makes_no_filesystem_changes(tmp_path: Path) -> None:
+    scandir = tmp_path / "run-service"; scandir.mkdir()
+    profile = _make_profile(tmp_path, "coder", state="running", with_pid=True)
+
+    actions = reconcile_profile_gateways(
+        hermes_home=tmp_path, scandir=scandir, dry_run=True,
+    )
+
+    # The action list is still produced...
+    assert _named_actions(actions) == [ReconcileAction(
+        profile="coder", prior_state="running", action="started",
+    )]
+    # ...but nothing on disk was touched.
+    assert (profile / "gateway.pid").exists()  # not removed under dry_run
+    assert not (scandir / "gateway-coder").exists()
+    assert not (tmp_path / "logs" / "container-boot.log").exists()
+
+
+def test_missing_profiles_root_still_registers_default_slot(
+    tmp_path: Path,
+) -> None:
+    """When $HERMES_HOME/profiles doesn't exist (fresh install), the
+    reconciliation should still register a gateway-default slot for
+    the root profile and return without raising. Previously this
+    returned an empty list; the default slot is now always present
+    so `hermes gateway start` (no -p) has somewhere to land."""
+    scandir = tmp_path / "run-service"; scandir.mkdir()
+    actions = reconcile_profile_gateways(
+        hermes_home=tmp_path, scandir=scandir, dry_run=False,
+    )
+    assert actions == [ReconcileAction(
+        profile="default", prior_state=None, action="registered",
+    )]
+    assert (scandir / "gateway-default").is_dir()
+    assert (scandir / "gateway-default" / "down").exists()
+
+
+def test_invalid_profile_name_in_directory_raises(tmp_path: Path) -> None:
+    """A profile dir whose name doesn't match validate_profile_name's
+    rules (uppercase, etc.) must surface as a hard error rather than
+    silently produce an invalid s6 service dir."""
+    scandir = tmp_path / "run-service"; scandir.mkdir()
+    _make_profile(tmp_path, "BadName", state="running")
+    with pytest.raises(ValueError):
+        reconcile_profile_gateways(
+            hermes_home=tmp_path, scandir=scandir, dry_run=False,
+        )
+
+
+def test_register_service_publishes_atomically(tmp_path: Path) -> None:
+    """The reconciler should build the new service dir in a sibling
+    tmp directory and rename it into place — never leaving a half-
+    populated slot visible to a concurrent s6-svscan rescan.
+
+    We verify the invariant indirectly: after a clean reconcile, the
+    target directory exists with all required files, and no sibling
+    .tmp leftovers remain. (Atomic publication is the only way to
+    achieve both with mkdir + write.)
+    """
+    scandir = tmp_path / "run-service"; scandir.mkdir()
+    _make_profile(tmp_path, "coder", state="running")
+
+    reconcile_profile_gateways(
+        hermes_home=tmp_path, scandir=scandir, dry_run=False,
+    )
+
+    # No leftover tmp dir.
+    leftover = list(scandir.glob("*.tmp"))
+    assert leftover == [], f"leftover tmp directories: {leftover}"
+
+    # Target is fully populated.
+    svc = scandir / "gateway-coder"
+    assert (svc / "type").exists()
+    assert (svc / "run").exists()
+    assert (svc / "log" / "run").exists()
+
+
+def test_register_service_overwrites_existing_slot(tmp_path: Path) -> None:
+    """A second reconciliation pass cleanly replaces an existing
+    slot (the tmp+rename publication overwrites the previous one)."""
+    scandir = tmp_path / "run-service"; scandir.mkdir()
+    profile = _make_profile(tmp_path, "coder", state="running")
+
+    # First pass.
+    reconcile_profile_gateways(
+        hermes_home=tmp_path, scandir=scandir, dry_run=False,
+    )
+    first_run = (scandir / "gateway-coder" / "run").read_text()
+
+    # Mutate the profile state so the run-script changes (extra_env
+    # rendering would differ if we wired profile config through, but
+    # for now just exercise the overwrite path).
+    (profile / "gateway_state.json").write_text(
+        '{"gateway_state": "stopped"}',
+    )
+    reconcile_profile_gateways(
+        hermes_home=tmp_path, scandir=scandir, dry_run=False,
+    )
+
+    # Slot still exists, no .tmp remnants.
+    assert (scandir / "gateway-coder" / "run").read_text() == first_run
+    assert list(scandir.glob("*.tmp")) == []
+    # Down marker now present (state went from running → stopped).
+    assert (scandir / "gateway-coder" / "down").exists()
+
+
+def test_register_service_cleans_up_stale_tmp_dir(tmp_path: Path) -> None:
+    """If a previous interrupted run left a .tmp sibling directory,
+    a fresh reconcile must clean it up rather than failing on mkdir."""
+    scandir = tmp_path / "run-service"; scandir.mkdir()
+    # Simulate a leftover from an interrupted run.
+    stale_tmp = scandir / "gateway-coder.tmp"
+    stale_tmp.mkdir()
+    (stale_tmp / "stale-file").write_text("garbage")
+
+    _make_profile(tmp_path, "coder", state="running")
+    reconcile_profile_gateways(
+        hermes_home=tmp_path, scandir=scandir, dry_run=False,
+    )
+
+    assert not stale_tmp.exists()
+    assert (scandir / "gateway-coder" / "run").exists()
+
+
+# ---------------------------------------------------------------------------
+# Default-profile slot — always registered (PR #30136 review item I1)
+# ---------------------------------------------------------------------------
+
+
+def test_default_slot_always_registered_on_empty_home(tmp_path: Path) -> None:
+    """Bare HERMES_HOME with nothing under it still produces a
+    gateway-default slot (down state)."""
+    scandir = tmp_path / "run-service"; scandir.mkdir()
+
+    actions = reconcile_profile_gateways(
+        hermes_home=tmp_path, scandir=scandir, dry_run=False,
+    )
+
+    assert actions == [ReconcileAction(
+        profile="default", prior_state=None, action="registered",
+    )]
+    svc = scandir / "gateway-default"
+    assert svc.is_dir()
+    assert (svc / "run").exists()
+    assert (svc / "down").exists()
+
+
+def test_default_slot_run_script_omits_profile_flag(tmp_path: Path) -> None:
+    """The default slot's run script must NOT pass `-p default` —
+    that would resolve to $HERMES_HOME/profiles/default/ instead of
+    the root profile. It must call `hermes gateway run` directly."""
+    scandir = tmp_path / "run-service"; scandir.mkdir()
+
+    reconcile_profile_gateways(
+        hermes_home=tmp_path, scandir=scandir, dry_run=False,
+    )
+
+    run = (scandir / "gateway-default" / "run").read_text()
+    assert "hermes gateway run" in run
+    assert "-p default" not in run
+    assert "-p 'default'" not in run
+
+
+def test_default_slot_autostarts_when_root_state_running(tmp_path: Path) -> None:
+    """gateway_state.json at the HERMES_HOME root with state=running
+    means the default slot auto-starts on container boot."""
+    scandir = tmp_path / "run-service"; scandir.mkdir()
+    _seed_default_root(tmp_path, state="running")
+
+    actions = reconcile_profile_gateways(
+        hermes_home=tmp_path, scandir=scandir, dry_run=False,
+    )
+
+    default_action = next(a for a in actions if a.profile == "default")
+    assert default_action.prior_state == "running"
+    assert default_action.action == "started"
+    assert not (scandir / "gateway-default" / "down").exists()
+
+
+def test_default_slot_does_not_autostart_when_root_state_stopped(
+    tmp_path: Path,
+) -> None:
+    scandir = tmp_path / "run-service"; scandir.mkdir()
+    _seed_default_root(tmp_path, state="stopped")
+
+    actions = reconcile_profile_gateways(
+        hermes_home=tmp_path, scandir=scandir, dry_run=False,
+    )
+
+    default_action = next(a for a in actions if a.profile == "default")
+    assert default_action.action == "registered"
+    assert (scandir / "gateway-default" / "down").exists()
+
+
+def test_default_slot_does_not_autostart_when_root_state_startup_failed(
+    tmp_path: Path,
+) -> None:
+    """Crash-loop guard applies to the default slot too."""
+    scandir = tmp_path / "run-service"; scandir.mkdir()
+    _seed_default_root(tmp_path, state="startup_failed")
+
+    actions = reconcile_profile_gateways(
+        hermes_home=tmp_path, scandir=scandir, dry_run=False,
+    )
+
+    default_action = next(a for a in actions if a.profile == "default")
+    assert default_action.action == "registered"
+
+
+def test_default_slot_cleans_up_stale_runtime_files_at_root(
+    tmp_path: Path,
+) -> None:
+    """gateway.pid and processes.json at the HERMES_HOME root (left
+    over from the previous container's default gateway) must be
+    swept the same way as for named profiles."""
+    scandir = tmp_path / "run-service"; scandir.mkdir()
+    _seed_default_root(tmp_path, state="running", with_pid=True)
+    assert (tmp_path / "gateway.pid").exists()
+
+    reconcile_profile_gateways(
+        hermes_home=tmp_path, scandir=scandir, dry_run=False,
+    )
+
+    assert not (tmp_path / "gateway.pid").exists()
+    assert not (tmp_path / "processes.json").exists()
+
+
+def test_default_slot_appears_before_named_profiles(tmp_path: Path) -> None:
+    """The action list is ordered: default first, then named profiles
+    in directory order. Operators and the boot-log reader rely on
+    this ordering being stable."""
+    scandir = tmp_path / "run-service"; scandir.mkdir()
+    _make_profile(tmp_path, "z-last-alphabetically", state="stopped")
+    _make_profile(tmp_path, "a-first-alphabetically", state="stopped")
+
+    actions = reconcile_profile_gateways(
+        hermes_home=tmp_path, scandir=scandir, dry_run=False,
+    )
+
+    assert [a.profile for a in actions] == [
+        "default",
+        "a-first-alphabetically",
+        "z-last-alphabetically",
+    ]
+
+
+def test_profiles_default_subdir_is_skipped_with_warning(
+    tmp_path: Path,
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    """A user-created profiles/default/ collides with the reserved
+    root-profile slot — the named entry is skipped (with a warning)
+    so we don't double-register gateway-default."""
+    import logging
+    caplog.set_level(logging.WARNING)
+    scandir = tmp_path / "run-service"; scandir.mkdir()
+    _make_profile(tmp_path, "default", state="running")
+
+    actions = reconcile_profile_gateways(
+        hermes_home=tmp_path, scandir=scandir, dry_run=False,
+    )
+
+    # Only the root-profile default slot appears — not the colliding
+    # named profile.
+    default_actions = [a for a in actions if a.profile == "default"]
+    assert len(default_actions) == 1
+    # And the warning surfaces so operators know the named profile
+    # was ignored.
+    assert any(
+        "profiles/default/" in record.message for record in caplog.records
+    )

tests/hermes_cli/test_gateway_s6_dispatch.py

@@ -0,0 +1,335 @@
+"""Tests for the Phase 4 s6 dispatch helper in hermes_cli.gateway.
+
+`_dispatch_via_service_manager_if_s6` decides whether a
+`hermes gateway start/stop/restart` invocation should be routed to
+the in-container S6ServiceManager instead of falling through to the
+host systemd/launchd/windows code path.
+"""
+from __future__ import annotations
+
+from typing import Any
+
+import pytest
+
+
+class _CallRecorder:
+    """Minimal stand-in for S6ServiceManager."""
+    kind = "s6"
+
+    def __init__(self) -> None:
+        self.calls: list[tuple[str, str]] = []
+
+    def start(self, name: str) -> None:
+        self.calls.append(("start", name))
+
+    def stop(self, name: str) -> None:
+        self.calls.append(("stop", name))
+
+    def restart(self, name: str) -> None:
+        self.calls.append(("restart", name))
+
+
+def test_dispatch_returns_false_on_host(monkeypatch: pytest.MonkeyPatch) -> None:
+    """When the environment isn't s6 (host run), the helper must
+    return False and not invoke a manager — callers continue with
+    their existing systemd/launchd/windows path."""
+    from hermes_cli import gateway as gw
+    monkeypatch.setattr(
+        "hermes_cli.service_manager.detect_service_manager", lambda: "systemd",
+    )
+    # Should not even attempt to construct a manager.
+    monkeypatch.setattr(
+        "hermes_cli.service_manager.get_service_manager",
+        lambda: pytest.fail("manager should not be constructed on host"),
+    )
+    assert gw._dispatch_via_service_manager_if_s6("start", profile="x") is False
+
+
+def test_dispatch_returns_true_and_calls_start_on_s6(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    from hermes_cli import gateway as gw
+    rec = _CallRecorder()
+    monkeypatch.setattr(
+        "hermes_cli.service_manager.detect_service_manager", lambda: "s6",
+    )
+    monkeypatch.setattr(
+        "hermes_cli.service_manager.get_service_manager", lambda: rec,
+    )
+    assert gw._dispatch_via_service_manager_if_s6("start", profile="coder") is True
+    assert rec.calls == [("start", "gateway-coder")]
+
+
+@pytest.mark.parametrize("action,expected", [
+    ("start", "start"),
+    ("stop", "stop"),
+    ("restart", "restart"),
+])
+def test_dispatch_translates_action_to_manager_method(
+    monkeypatch: pytest.MonkeyPatch, action: str, expected: str,
+) -> None:
+    from hermes_cli import gateway as gw
+    rec = _CallRecorder()
+    monkeypatch.setattr(
+        "hermes_cli.service_manager.detect_service_manager", lambda: "s6",
+    )
+    monkeypatch.setattr(
+        "hermes_cli.service_manager.get_service_manager", lambda: rec,
+    )
+    assert gw._dispatch_via_service_manager_if_s6(action, profile="x") is True
+    assert rec.calls == [(expected, "gateway-x")]
+
+
+def test_dispatch_unknown_action_returns_false(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """An unrecognized action (e.g. 'install') must not silently
+    succeed — return False so the host code path handles it."""
+    from hermes_cli import gateway as gw
+    rec = _CallRecorder()
+    monkeypatch.setattr(
+        "hermes_cli.service_manager.detect_service_manager", lambda: "s6",
+    )
+    monkeypatch.setattr(
+        "hermes_cli.service_manager.get_service_manager", lambda: rec,
+    )
+    assert gw._dispatch_via_service_manager_if_s6("install", profile="x") is False
+    assert rec.calls == []
+
+
+def test_dispatch_defaults_profile_to_default(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """When profile is None, the helper resolves it via _profile_arg().
+    With no profile context set anywhere, that resolves to "default"."""
+    from hermes_cli import gateway as gw
+    rec = _CallRecorder()
+    monkeypatch.setattr(
+        "hermes_cli.service_manager.detect_service_manager", lambda: "s6",
+    )
+    monkeypatch.setattr(
+        "hermes_cli.service_manager.get_service_manager", lambda: rec,
+    )
+    monkeypatch.setattr(
+        "hermes_cli.gateway._profile_suffix", lambda: "",
+    )
+    assert gw._dispatch_via_service_manager_if_s6("start") is True
+    assert rec.calls == [("start", "gateway-default")]
+
+
+# ---------------------------------------------------------------------------
+# _dispatch_all_via_service_manager_if_s6 — --all under s6
+# ---------------------------------------------------------------------------
+
+
+class _ListingRecorder(_CallRecorder):
+    """_CallRecorder that also exposes a profile list."""
+
+    def __init__(self, profiles: list[str]) -> None:
+        super().__init__()
+        self._profiles = profiles
+
+    def list_profile_gateways(self) -> list[str]:
+        return list(self._profiles)
+
+
+def test_dispatch_all_returns_false_on_host(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    from hermes_cli import gateway as gw
+    monkeypatch.setattr(
+        "hermes_cli.service_manager.detect_service_manager", lambda: "systemd",
+    )
+    monkeypatch.setattr(
+        "hermes_cli.service_manager.get_service_manager",
+        lambda: pytest.fail("manager should not be constructed on host"),
+    )
+    assert gw._dispatch_all_via_service_manager_if_s6("stop") is False
+
+
+def test_dispatch_all_iterates_every_profile_on_stop(
+    monkeypatch: pytest.MonkeyPatch,
+    capsys: pytest.CaptureFixture,
+) -> None:
+    from hermes_cli import gateway as gw
+    rec = _ListingRecorder(["coder", "writer", "assistant"])
+    monkeypatch.setattr(
+        "hermes_cli.service_manager.detect_service_manager", lambda: "s6",
+    )
+    monkeypatch.setattr(
+        "hermes_cli.service_manager.get_service_manager", lambda: rec,
+    )
+    assert gw._dispatch_all_via_service_manager_if_s6("stop") is True
+    assert rec.calls == [
+        ("stop", "gateway-coder"),
+        ("stop", "gateway-writer"),
+        ("stop", "gateway-assistant"),
+    ]
+    out = capsys.readouterr().out
+    assert "Stopped 3 profile gateway(s)" in out
+
+
+def test_dispatch_all_iterates_every_profile_on_restart(
+    monkeypatch: pytest.MonkeyPatch,
+    capsys: pytest.CaptureFixture,
+) -> None:
+    from hermes_cli import gateway as gw
+    rec = _ListingRecorder(["coder", "writer"])
+    monkeypatch.setattr(
+        "hermes_cli.service_manager.detect_service_manager", lambda: "s6",
+    )
+    monkeypatch.setattr(
+        "hermes_cli.service_manager.get_service_manager", lambda: rec,
+    )
+    assert gw._dispatch_all_via_service_manager_if_s6("restart") is True
+    assert rec.calls == [
+        ("restart", "gateway-coder"),
+        ("restart", "gateway-writer"),
+    ]
+    out = capsys.readouterr().out
+    assert "Restarted 2 profile gateway(s)" in out
+
+
+def test_dispatch_all_handles_partial_failure(
+    monkeypatch: pytest.MonkeyPatch,
+    capsys: pytest.CaptureFixture,
+) -> None:
+    """A failure on one profile must not skip the others; the helper
+    reports each failure and the success count."""
+    from hermes_cli import gateway as gw
+
+    class _FailOnWriter(_ListingRecorder):
+        def stop(self, name: str) -> None:
+            if name == "gateway-writer":
+                raise RuntimeError("supervise FIFO permission denied")
+            super().stop(name)
+
+    rec = _FailOnWriter(["coder", "writer", "assistant"])
+    monkeypatch.setattr(
+        "hermes_cli.service_manager.detect_service_manager", lambda: "s6",
+    )
+    monkeypatch.setattr(
+        "hermes_cli.service_manager.get_service_manager", lambda: rec,
+    )
+    assert gw._dispatch_all_via_service_manager_if_s6("stop") is True
+    # The two successful ones were called; writer raised before recording.
+    assert ("stop", "gateway-coder") in rec.calls
+    assert ("stop", "gateway-assistant") in rec.calls
+    assert ("stop", "gateway-writer") not in rec.calls
+    out = capsys.readouterr().out
+    assert "Stopped 2 profile gateway(s)" in out
+    assert "Could not stop gateway-writer" in out
+    assert "supervise FIFO permission denied" in out
+
+
+def test_dispatch_all_empty_list_reports_and_returns_true(
+    monkeypatch: pytest.MonkeyPatch,
+    capsys: pytest.CaptureFixture,
+) -> None:
+    """With no profile gateways registered the helper still claims the
+    dispatch (returns True) and prints a friendly message — the host
+    fallback would just pkill nothing, which isn't useful inside a
+    container."""
+    from hermes_cli import gateway as gw
+    rec = _ListingRecorder([])
+    monkeypatch.setattr(
+        "hermes_cli.service_manager.detect_service_manager", lambda: "s6",
+    )
+    monkeypatch.setattr(
+        "hermes_cli.service_manager.get_service_manager", lambda: rec,
+    )
+    assert gw._dispatch_all_via_service_manager_if_s6("stop") is True
+    assert rec.calls == []
+    assert "No profile gateways" in capsys.readouterr().out
+
+
+def test_dispatch_all_unknown_action_returns_false(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """`start --all` is not a supported CLI surface; the helper must
+    fall through to the host code path rather than no-op."""
+    from hermes_cli import gateway as gw
+    monkeypatch.setattr(
+        "hermes_cli.service_manager.detect_service_manager", lambda: "s6",
+    )
+    monkeypatch.setattr(
+        "hermes_cli.service_manager.get_service_manager",
+        lambda: pytest.fail(
+            "manager should not be constructed for unsupported --all action",
+        ),
+    )
+    assert gw._dispatch_all_via_service_manager_if_s6("start") is False
+
+
+# ---------------------------------------------------------------------------
+# Friendly error rendering — GatewayNotRegisteredError / S6CommandError
+# (PR #30136 review item I2)
+# ---------------------------------------------------------------------------
+
+
+def test_dispatch_renders_gateway_not_registered_friendly(
+    monkeypatch: pytest.MonkeyPatch,
+    capsys: pytest.CaptureFixture,
+) -> None:
+    """`hermes -p typo gateway start` should print a clear message and
+    exit 1 — not dump a traceback at the user."""
+    from hermes_cli import gateway as gw
+    from hermes_cli.service_manager import GatewayNotRegisteredError
+
+    class _RaisesMissing:
+        kind = "s6"
+
+        def start(self, name: str) -> None:
+            raise GatewayNotRegisteredError("typo")
+
+    monkeypatch.setattr(
+        "hermes_cli.service_manager.detect_service_manager", lambda: "s6",
+    )
+    monkeypatch.setattr(
+        "hermes_cli.service_manager.get_service_manager", lambda: _RaisesMissing(),
+    )
+
+    with pytest.raises(SystemExit) as excinfo:
+        gw._dispatch_via_service_manager_if_s6("start", profile="typo")
+    assert excinfo.value.code == 1
+    out = capsys.readouterr().out
+    assert "no such gateway 'typo'" in out
+    assert "hermes profile create typo" in out
+    # And critically: no traceback prefix.
+    assert "Traceback" not in out
+
+
+def test_dispatch_renders_s6_command_error_friendly(
+    monkeypatch: pytest.MonkeyPatch,
+    capsys: pytest.CaptureFixture,
+) -> None:
+    """An s6-svc failure (e.g. EACCES on the supervise FIFO) should
+    surface the stderr inline, not as an opaque traceback."""
+    from hermes_cli import gateway as gw
+    from hermes_cli.service_manager import S6CommandError
+
+    class _RaisesS6Error:
+        kind = "s6"
+
+        def start(self, name: str) -> None:
+            raise S6CommandError(
+                service=name,
+                action="start",
+                returncode=111,
+                stderr="s6-svc: fatal: Permission denied",
+            )
+
+    monkeypatch.setattr(
+        "hermes_cli.service_manager.detect_service_manager", lambda: "s6",
+    )
+    monkeypatch.setattr(
+        "hermes_cli.service_manager.get_service_manager", lambda: _RaisesS6Error(),
+    )
+
+    with pytest.raises(SystemExit) as excinfo:
+        gw._dispatch_via_service_manager_if_s6("start", profile="coder")
+    assert excinfo.value.code == 1
+    out = capsys.readouterr().out
+    assert "rc=111" in out
+    assert "Permission denied" in out
+    assert "Traceback" not in out

tests/hermes_cli/test_profiles_s6_hooks.py

@@ -0,0 +1,210 @@
+"""Tests for the Phase 4 s6 hooks in hermes_cli.profiles.
+
+Specifically: _maybe_register_gateway_service,
+_maybe_unregister_gateway_service. The integration with
+create_profile and delete_profile is covered indirectly by the
+existing TestCreateProfile and TestDeleteProfile classes in
+tests/hermes_cli/test_profiles.py; here we only exercise the new
+helper surface that doesn't touch the filesystem.
+"""
+from __future__ import annotations
+
+from typing import Any
+
+import pytest
+
+from hermes_cli.profiles import (
+    _maybe_register_gateway_service,
+    _maybe_unregister_gateway_service,
+)
+
+
+# ---------------------------------------------------------------------------
+# _maybe_register_gateway_service / _maybe_unregister_gateway_service
+# ---------------------------------------------------------------------------
+
+
+class _HostManager:
+    """Mimics a host backend that doesn't support runtime registration."""
+    kind = "systemd"
+
+    def supports_runtime_registration(self) -> bool:
+        return False
+
+    def register_profile_gateway(self, *args: Any, **kwargs: Any) -> None:
+        raise AssertionError("host backend register_profile_gateway should not be called")
+
+    def unregister_profile_gateway(self, *args: Any, **kwargs: Any) -> None:
+        raise AssertionError("host backend unregister_profile_gateway should not be called")
+
+
+class _S6Manager:
+    """Mimics S6ServiceManager just enough for the hooks."""
+    kind = "s6"
+
+    def __init__(self) -> None:
+        self.registered: list[str] = []
+        self.unregistered: list[str] = []
+        self.raise_on_register: Exception | None = None
+        self.raise_on_unregister: Exception | None = None
+
+    def supports_runtime_registration(self) -> bool:
+        return True
+
+    def register_profile_gateway(
+        self, profile: str, *,
+        extra_env: dict[str, str] | None = None,
+    ) -> None:
+        if self.raise_on_register is not None:
+            raise self.raise_on_register
+        self.registered.append(profile)
+
+    def unregister_profile_gateway(self, profile: str) -> None:
+        if self.raise_on_unregister is not None:
+            raise self.raise_on_unregister
+        self.unregistered.append(profile)
+
+
+def _patch_detect_s6(monkeypatch: pytest.MonkeyPatch) -> None:
+    """Pretend we're inside an s6 container so the host short-circuit
+    in :func:`_maybe_register_gateway_service` /
+    :func:`_maybe_unregister_gateway_service` doesn't fire.
+
+    Without this, ``detect_service_manager()`` runs its real
+    implementation (host Linux/macOS in CI), returns ``"systemd"`` or
+    ``"launchd"``, and the hooks return early before reaching the
+    patched ``get_service_manager``. Each s6-call-through test
+    explicitly opts into this so the host-no-op tests can still
+    exercise the early-return path.
+    """
+    monkeypatch.setattr(
+        "hermes_cli.service_manager.detect_service_manager",
+        lambda: "s6",
+    )
+
+
+def test_register_noop_on_host(monkeypatch: pytest.MonkeyPatch) -> None:
+    # NOTE: deliberately DO NOT patch detect_service_manager — we want
+    # the real host detection to kick in and short-circuit before
+    # get_service_manager is ever called. The lambda below is a
+    # defense-in-depth assertion that get_service_manager is never
+    # reached on host.
+    monkeypatch.setattr(
+        "hermes_cli.service_manager.get_service_manager",
+        lambda: _HostManager(),
+    )
+    # Should NOT raise the AssertionError from _HostManager.register
+    _maybe_register_gateway_service("hostprof")
+
+
+def test_register_calls_through_on_s6(monkeypatch: pytest.MonkeyPatch) -> None:
+    _patch_detect_s6(monkeypatch)
+    mgr = _S6Manager()
+    monkeypatch.setattr(
+        "hermes_cli.service_manager.get_service_manager", lambda: mgr,
+    )
+    _maybe_register_gateway_service("coder")
+    assert mgr.registered == ["coder"]
+
+
+def test_register_swallows_duplicate_value_error(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """A pre-existing s6 registration (from container-boot reconcile)
+    is a benign condition — register must not propagate ValueError."""
+    _patch_detect_s6(monkeypatch)
+    mgr = _S6Manager()
+    mgr.raise_on_register = ValueError("already registered")
+    monkeypatch.setattr(
+        "hermes_cli.service_manager.get_service_manager", lambda: mgr,
+    )
+    # Should NOT raise
+    _maybe_register_gateway_service("coder")
+
+
+def test_register_swallows_arbitrary_error(
+    monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str],
+) -> None:
+    """Even an unexpected exception from the manager must not bring
+    down `hermes profile create` — print and continue."""
+    _patch_detect_s6(monkeypatch)
+    mgr = _S6Manager()
+    mgr.raise_on_register = RuntimeError("svscanctl exploded")
+    monkeypatch.setattr(
+        "hermes_cli.service_manager.get_service_manager", lambda: mgr,
+    )
+    _maybe_register_gateway_service("coder")
+    captured = capsys.readouterr()
+    assert "Could not register" in captured.out
+
+
+def test_register_swallows_no_backend_runtime_error(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """When `get_service_manager()` raises RuntimeError (no backend
+    detected), the hook must silently no-op."""
+    _patch_detect_s6(monkeypatch)
+    def _no_backend() -> None:
+        raise RuntimeError("no supported service manager detected")
+    monkeypatch.setattr(
+        "hermes_cli.service_manager.get_service_manager", _no_backend,
+    )
+    # Should NOT raise
+    _maybe_register_gateway_service("anywhere")
+
+
+def test_register_silent_when_detect_throws(
+    monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str],
+) -> None:
+    """If detect_service_manager itself raises (e.g. a partial s6
+    install on a host machine), the hook must stay silent — no
+    confusing s6 warning printed to a user who has never touched a
+    container."""
+    def _broken_detect() -> str:
+        raise RuntimeError("detection blew up")
+    monkeypatch.setattr(
+        "hermes_cli.service_manager.detect_service_manager", _broken_detect,
+    )
+    # If get_service_manager is reached, the test will assert via
+    # _HostManager.register. It must NOT be reached.
+    monkeypatch.setattr(
+        "hermes_cli.service_manager.get_service_manager",
+        lambda: _HostManager(),
+    )
+    _maybe_register_gateway_service("anywhere")
+    captured = capsys.readouterr()
+    assert "Could not register" not in captured.out
+    assert captured.out == ""
+
+
+def test_unregister_noop_on_host(monkeypatch: pytest.MonkeyPatch) -> None:
+    # Same as test_register_noop_on_host: rely on real host detection.
+    monkeypatch.setattr(
+        "hermes_cli.service_manager.get_service_manager",
+        lambda: _HostManager(),
+    )
+    _maybe_unregister_gateway_service("hostprof")
+
+
+def test_unregister_calls_through_on_s6(monkeypatch: pytest.MonkeyPatch) -> None:
+    _patch_detect_s6(monkeypatch)
+    mgr = _S6Manager()
+    monkeypatch.setattr(
+        "hermes_cli.service_manager.get_service_manager", lambda: mgr,
+    )
+    _maybe_unregister_gateway_service("coder")
+    assert mgr.unregistered == ["coder"]
+
+
+def test_unregister_swallows_errors(
+    monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str],
+) -> None:
+    _patch_detect_s6(monkeypatch)
+    mgr = _S6Manager()
+    mgr.raise_on_unregister = RuntimeError("svc gone weird")
+    monkeypatch.setattr(
+        "hermes_cli.service_manager.get_service_manager", lambda: mgr,
+    )
+    _maybe_unregister_gateway_service("coder")
+    captured = capsys.readouterr()
+    assert "Could not unregister" in captured.out

tests/hermes_cli/test_service_manager.py

@@ -0,0 +1,793 @@
+"""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")
+
+
+# ---------------------------------------------------------------------------
+# _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."""
+    from hermes_cli.service_manager import S6ServiceManager
+    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:
+    from hermes_cli.service_manager import S6ServiceManager
+    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:
+    from hermes_cli.service_manager import S6ServiceManager
+    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 "hermes -p coder gateway run" in run_text
+    assert "s6-setuidgid hermes" 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"
+    )
+
+    # 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_extra_env_is_quoted(s6_scandir, fake_subprocess_run) -> None:
+    from hermes_cli.service_manager import S6ServiceManager
+    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_s6_register_rejects_invalid_profile_name(s6_scandir) -> None:
+    from hermes_cli.service_manager import S6ServiceManager
+    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:
+    from hermes_cli.service_manager import S6ServiceManager
+    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
+    from hermes_cli.service_manager import S6ServiceManager
+
+    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:
+    from hermes_cli.service_manager import S6ServiceManager
+    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:
+    from hermes_cli.service_manager import S6ServiceManager
+    # 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:
+    from hermes_cli.service_manager import S6ServiceManager
+    # 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:
+    from hermes_cli.service_manager import S6ServiceManager
+    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:
+    from hermes_cli.service_manager import S6ServiceManager
+    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"]
+
+
+# ---------------------------------------------------------------------------
+# 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,
+        S6ServiceManager,
+    )
+
+    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,
+        S6ServiceManager,
+    )
+
+    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,
+        S6ServiceManager,
+    )
+
+    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, S6ServiceManager
+
+    # 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
+    from hermes_cli.service_manager import S6ServiceManager
+
+    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

tests/tools/test_docker_environment.py

@@ -385,18 +385,19 @@ def test_normalize_env_dict_rejects_complex_values():
     assert result == {"GOOD": "string"}
 
 
-def test_security_args_include_setuid_setgid_for_gosu_drop(monkeypatch):
+def test_security_args_include_setuid_setgid_for_privdrop(monkeypatch):
     """The default (run_as_host_user=False) invocation must include SETUID and
-    SETGID caps so the image entrypoint can drop from root to the non-root
-    `hermes` user via gosu.
+    SETGID caps so the image's init can drop from root to a non-root user
+    (e.g. via ``s6-setuidgid`` in the bundled Hermes image, or ``gosu``/``su``
+    in user-provided images).
 
-    Without these caps gosu exits with
-    ``error: failed switching to 'hermes': operation not permitted``
-    and the container exits immediately (exit 1) before running any work.
+    Without these caps the privilege-drop helper fails with
+    ``operation not permitted`` and the container exits immediately (exit 1)
+    before running any work.
 
-    `no-new-privileges` is kept, so gosu still cannot escalate back to root
-    after the drop — the drop is a one-way transition performed before the
-    `no_new_privs` bit is enforced on the exec boundary.
+    ``no-new-privileges`` is kept, so the dropped process still cannot
+    escalate back to root after the drop — the drop is a one-way transition
+    performed before the ``no_new_privs`` bit is enforced on the exec boundary.
     """
     monkeypatch.setattr(docker_env, "find_docker", lambda: "/usr/bin/docker")
     calls = _mock_subprocess_run(monkeypatch)
@@ -412,8 +413,8 @@ def test_security_args_include_setuid_setgid_for_gosu_drop(monkeypatch):
         for i, flag in enumerate(run_args[:-1])
         if flag == "--cap-add"
     }
-    assert "SETUID" in added, "SETUID cap missing — gosu drop in entrypoint will fail"
-    assert "SETGID" in added, "SETGID cap missing — gosu drop in entrypoint will fail"
+    assert "SETUID" in added, "SETUID cap missing — image privilege-drop will fail"
+    assert "SETGID" in added, "SETGID cap missing — image privilege-drop will fail"
 
 
 # ── run_as_host_user tests ────────────────────────────────────────
@@ -441,8 +442,9 @@ def test_run_as_host_user_passes_uid_gid(monkeypatch):
 
 
 def test_run_as_host_user_drops_setuid_setgid_caps(monkeypatch):
-    """When --user is passed, the container never needs gosu, so SETUID/SETGID
-    caps are omitted for a tighter security posture."""
+    """When --user is passed, the container already starts unprivileged and
+    never needs a privilege drop, so SETUID/SETGID caps are omitted for a
+    tighter security posture."""
     monkeypatch.setattr(docker_env, "find_docker", lambda: "/usr/bin/docker")
     monkeypatch.setattr(docker_env.os, "getuid", lambda: 1000, raising=False)
     monkeypatch.setattr(docker_env.os, "getgid", lambda: 1000, raising=False)
@@ -459,10 +461,10 @@ def test_run_as_host_user_drops_setuid_setgid_caps(monkeypatch):
         if flag == "--cap-add"
     }
     assert "SETUID" not in added, (
-        "SETUID cap should be dropped when running as host user — no gosu drop is needed"
+        "SETUID cap should be dropped when running as host user — no privilege drop is needed"
     )
     assert "SETGID" not in added, (
-        "SETGID cap should be dropped when running as host user — no gosu drop is needed"
+        "SETGID cap should be dropped when running as host user — no privilege drop is needed"
     )
     # Core non-privilege-drop caps must still be there (pip/npm/apt need them).
     assert "DAC_OVERRIDE" in added

tests/tools/test_dockerfile_pid1_reaping.py

@@ -5,11 +5,17 @@ they deliberately avoid snapshotting specific package versions, line numbers,
 or exact flag choices.  What they DO assert is that the Dockerfile maintains
 the properties required for correct production behaviour:
 
-- A PID-1 init (tini) is installed and wraps the entrypoint, so that orphaned
+- A PID-1 init is installed and wraps the entrypoint, so that orphaned
   subprocesses (MCP stdio servers, git, bun, browser daemons) get reaped
   instead of accumulating as zombies (#15012).
 - Signal forwarding runs through the init so ``docker stop`` triggers
   hermes's own graceful-shutdown path.
+
+The init can be any reaper-capable PID-1: the historical lineage was
+``tini``; the current image uses s6-overlay's ``/init`` (which execs
+``s6-svscan`` as PID 1, with the same SIGCHLD-reaping property). The
+checks below accept either family — the contract is behavioural, not
+nominal.
 """
 
 from __future__ import annotations
@@ -24,6 +30,21 @@ DOCKERFILE = REPO_ROOT / "Dockerfile"
 DOCKERIGNORE = REPO_ROOT / ".dockerignore"
 
 
+# Init-process families this repo accepts as PID 1. ``tini`` /
+# ``dumb-init`` / ``catatonit`` are classic minimal reapers; s6-overlay
+# ships ``/init`` which execs ``s6-svscan`` as PID 1 (same reaper
+# contract, plus supervision of declared services). Either family
+# satisfies the zombie-reaping invariant — see issue #15012.
+_KNOWN_INIT_TOKENS: tuple[str, ...] = (
+    "tini",
+    "dumb-init",
+    "catatonit",
+    "s6-overlay",
+    "s6-svscan",
+    "/init",
+)
+
+
 @pytest.fixture(scope="module")
 def dockerfile_text() -> str:
     if not DOCKERFILE.exists():
@@ -57,8 +78,17 @@ def _run_steps(dockerfile_text: str) -> list[str]:
     ]
 
 
+def _instruction_text(dockerfile_text: str) -> str:
+    """Join every non-comment Dockerfile instruction into one searchable
+    string. Crucially excludes comments — otherwise the historical
+    explanation of "we used to use tini" would silently satisfy a
+    substring check long after tini was removed from the build.
+    """
+    return "\n".join(_dockerfile_instructions(dockerfile_text))
+
+
 def test_dockerfile_installs_an_init_for_zombie_reaping(dockerfile_text):
-    """Some init (tini, dumb-init, catatonit) must be installed.
+    """Some init (tini, dumb-init, catatonit, s6-overlay) must be installed.
 
     Without a PID-1 init that handles SIGCHLD, hermes accumulates zombie
     processes from MCP stdio subprocesses, git operations, browser
@@ -67,12 +97,17 @@ def test_dockerfile_installs_an_init_for_zombie_reaping(dockerfile_text):
     """
     # Accept any of the common reapers.  The contract is behavioural:
     # something must be installed that reaps orphans.
-    known_inits = ("tini", "dumb-init", "catatonit")
-    installed = any(name in dockerfile_text for name in known_inits)
+    #
+    # Scan instructions only (no comments) so a stale historical mention
+    # in a comment can't masquerade as a current install. Without this,
+    # removing tini from the actual build but leaving the word in a
+    # comment would silently keep the test green.
+    instructions = _instruction_text(dockerfile_text)
+    installed = any(name in instructions for name in _KNOWN_INIT_TOKENS)
     assert installed, (
-        "No PID-1 init detected in Dockerfile (looked for: "
-        f"{', '.join(known_inits)}). Without an init process to reap "
-        "orphaned subprocesses, hermes accumulates zombies in Docker "
+        "No PID-1 init detected in Dockerfile instructions (looked for: "
+        f"{', '.join(_KNOWN_INIT_TOKENS)}). Without an init process to "
+        "reap orphaned subprocesses, hermes accumulates zombies in Docker "
         "deployments. See issue #15012."
     )
 
@@ -80,8 +115,8 @@ def test_dockerfile_installs_an_init_for_zombie_reaping(dockerfile_text):
 def test_dockerfile_entrypoint_routes_through_the_init(dockerfile_text):
     """The ENTRYPOINT must invoke the init, not the entrypoint script directly.
 
-    Installing tini is only half the fix — the container must actually run
-    with tini as PID 1.  If the ENTRYPOINT executes the shell script
+    Installing the init is only half the fix — the container must actually
+    run with it as PID 1.  If the ENTRYPOINT executes the shell script
     directly, the shell becomes PID 1 and will ``exec`` into hermes,
     which then runs as PID 1 without any zombie reaping.
     """
@@ -96,12 +131,12 @@ def test_dockerfile_entrypoint_routes_through_the_init(dockerfile_text):
 
     assert entrypoint_line is not None, "Dockerfile is missing an ENTRYPOINT directive"
 
-    known_inits = ("tini", "dumb-init", "catatonit")
-    routes_through_init = any(name in entrypoint_line for name in known_inits)
+    routes_through_init = any(name in entrypoint_line for name in _KNOWN_INIT_TOKENS)
     assert routes_through_init, (
-        f"ENTRYPOINT does not route through an init: {entrypoint_line!r}. "
-        "If tini is only installed but not wired into ENTRYPOINT, hermes "
-        "still runs as PID 1 and zombies will accumulate (#15012)."
+        f"ENTRYPOINT does not route through a PID-1 init: {entrypoint_line!r}. "
+        f"Expected one of {_KNOWN_INIT_TOKENS}. If the init is installed but "
+        "not wired into ENTRYPOINT, hermes still runs as PID 1 and zombies "
+        "will accumulate (#15012)."
     )