"""Managed-gateway isolation for dispatcher-owned Kanban workers.""" from __future__ import annotations import json import subprocess import sys import time from pathlib import Path import pytest from hermes_cli import kanban_db as kb from hermes_cli import kanban_db_dispatch as kbd @pytest.fixture def worker_setup(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> tuple[Path, kb.Task]: root = tmp_path / ".hermes" profile = root / "profiles" / "coder" profile.mkdir(parents=True) root.joinpath("config.yaml").write_text("{}\n", encoding="utf-8") profile.joinpath("config.yaml").write_text("{}\n", encoding="utf-8") monkeypatch.setenv("HOME", str(tmp_path)) monkeypatch.setenv("HERMES_HOME", str(root)) monkeypatch.setattr(Path, "home", lambda: tmp_path) workspace = tmp_path / "candidate-worktree" workspace.mkdir() task = kb.Task( id="t_candidate_restart", title="activate candidate", body=None, assignee="coder", status="running", priority=0, created_by="test", created_at=1, started_at=1, completed_at=None, workspace_kind="worktree", workspace_path=str(workspace), claim_lock="host:dispatcher", claim_expires=999, tenant=None, branch_name="wt/t_candidate_restart", current_run_id=23, ) return workspace, task @pytest.mark.platforms("linux") def test_managed_gateway_worker_is_spawned_in_restart_safe_scope( worker_setup: tuple[Path, kb.Task], monkeypatch: pytest.MonkeyPatch ) -> None: workspace, task = worker_setup captured_cmd: list[str] = [] captured_env: dict[str, str] = {} captured_cwd: str | None = None class FakeProc: pid = 4242 def fake_popen(cmd, **kwargs): nonlocal captured_cwd captured_cmd.extend(cmd) captured_env.update(kwargs.get("env") or {}) captured_cwd = kwargs.get("cwd") return FakeProc() monkeypatch.setenv("INVOCATION_ID", "managed-gateway-test") monkeypatch.setenv("ANTHROPIC_API_KEY", "must-not-cross-profile") monkeypatch.setattr("agent.secret_scope.is_multiplex_active", lambda: True) monkeypatch.setattr(subprocess, "Popen", fake_popen) monkeypatch.setattr("tools.process_registry._is_supervised_gateway_process", lambda: True) monkeypatch.setattr("tools.process_registry._systemd_run_user_scope_available", lambda: True) monkeypatch.setattr("tools.process_registry._worker_memory_max_bytes", lambda: 536_870_912) monkeypatch.setattr("shutil.which", lambda name: "/usr/bin/systemd-run") assert kbd._default_spawn(task, str(workspace)) == 4242 assert captured_cmd[:4] == ["/usr/bin/systemd-run", "--user", "--scope", "--quiet"] unit_index = captured_cmd.index("--unit") assert captured_cmd[unit_index + 1] == "hermes-worker-kanban-t_candidate_restart-run-23" assert "MemoryMax=536870912" in captured_cmd separator = captured_cmd.index("--") assert captured_cmd[separator + 1 :] == [sys.executable, "-m", "hermes_cli.kanban_worker_client"] assert task.assignee is not None assert captured_env["HERMES_PROFILE"] == task.assignee assert captured_env["HERMES_HOME"] == str(workspace.parent / ".hermes" / "profiles" / task.assignee) assert captured_cwd == str(workspace) assert captured_env["HERMES_KANBAN_TASK"] == task.id assert captured_env["HERMES_KANBAN_RUN_ID"] == "23" assert "ANTHROPIC_API_KEY" not in captured_env @pytest.mark.platforms("linux") def test_managed_gateway_worker_spawn_fails_closed_without_scope( worker_setup: tuple[Path, kb.Task], monkeypatch: pytest.MonkeyPatch ) -> None: workspace, task = worker_setup popen_calls: list[list[str]] = [] monkeypatch.setenv("INVOCATION_ID", "managed-gateway-test") monkeypatch.setattr(subprocess, "Popen", lambda cmd, **kwargs: popen_calls.append(list(cmd))) monkeypatch.setattr("tools.process_registry._is_supervised_gateway_process", lambda: True) monkeypatch.setattr("tools.process_registry._systemd_run_user_scope_available", lambda: False) with pytest.raises(RuntimeError, match="restart-safe systemd scope"): kbd._default_spawn(task, str(workspace)) assert popen_calls == [] @pytest.mark.platforms("linux") def test_managed_gateway_scope_builder_fails_closed_if_binary_disappears( worker_setup: tuple[Path, kb.Task], monkeypatch: pytest.MonkeyPatch ) -> None: workspace, task = worker_setup monkeypatch.setenv("INVOCATION_ID", "managed-gateway-test") monkeypatch.setattr("tools.process_registry._is_supervised_gateway_process", lambda: True) monkeypatch.setattr("tools.process_registry._systemd_run_user_scope_available", lambda: True) monkeypatch.setattr("shutil.which", lambda _name: None) monkeypatch.setattr(subprocess, "Popen", lambda *_args, **_kwargs: pytest.fail("unsafe direct spawn")) with pytest.raises(RuntimeError, match="restart-safe systemd scope"): kbd._default_spawn(task, str(workspace)) def test_standalone_dispatcher_keeps_direct_worker_spawn( worker_setup: tuple[Path, kb.Task], monkeypatch: pytest.MonkeyPatch ) -> None: workspace, task = worker_setup captured_cmd: list[str] = [] class FakeProc: pid = 4243 # Standalone = no systemd unit at all (CI runners inherit INVOCATION_ID). monkeypatch.delenv("INVOCATION_ID", raising=False) monkeypatch.setattr(subprocess, "Popen", lambda cmd, **kwargs: captured_cmd.extend(cmd) or FakeProc()) monkeypatch.setattr("tools.process_registry._is_supervised_gateway_process", lambda: False) monkeypatch.setattr( "tools.process_registry._systemd_run_user_scope_available", lambda: pytest.fail("scope probe must not run outside managed gateway"), ) assert kbd._default_spawn(task, str(workspace)) == 4243 assert captured_cmd == [sys.executable, "-m", "hermes_cli.kanban_worker_client"] @pytest.mark.platforms("linux") def test_oneshot_unit_dispatcher_scope_wraps_or_warns_never_dooms_silently( worker_setup: tuple[Path, kb.Task], monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture ) -> None: """A dispatcher under any systemd unit that is NOT the supervised gateway (a ``Type=oneshot`` dispatch timer, #113612) loses its cgroup at unit exit. With a user bus the worker gets its own ``hermes-worker-*`` scope; without one the spawn still happens (the unit's ``KillMode``/lifetime is unknowable, a long-lived sequencer keeps working) but the operator is told exactly why the workers may die. A cron caller under the same unit is unchanged.""" from tools import process_registry workspace, task = worker_setup spawned: list[list[str]] = [] class FakeProc: pid = 4243 monkeypatch.setattr(subprocess, "Popen", lambda cmd, **kwargs: spawned.append(list(cmd)) or FakeProc()) monkeypatch.setenv("INVOCATION_ID", "oneshot-dispatch-timer") monkeypatch.setattr(process_registry, "_is_supervised_gateway_process", lambda: False) monkeypatch.setattr(process_registry, "_systemd_run_user_scope_available", lambda: True) monkeypatch.setattr( process_registry, "_build_systemd_scope_argv", lambda cmd, unit_suffix: ["systemd-run", "--user", "--scope", "--unit", f"hermes-worker-{unit_suffix}", *cmd], ) kbd._default_spawn(task, str(workspace)) assert spawned[-1][:3] == ["systemd-run", "--user", "--scope"] assert f"hermes-worker-kanban-{task.id}-run-{task.current_run_id}" in spawned[-1] monkeypatch.setattr(process_registry, "_systemd_run_user_scope_available", lambda: False) monkeypatch.setattr(process_registry, "_scope_degraded_warned", False) with caplog.at_level("WARNING", logger=process_registry.logger.name): kbd._default_spawn(task, str(workspace)) # Unwrapped: the bare worker argv (the profile owner executes; this client waits for its receipt). assert spawned[-1] == [sys.executable, "-m", "hermes_cli.kanban_worker_client"] warned = [r.getMessage() for r in caplog.records if "KILLED when the unit exits" in r.getMessage()] assert len(warned) == 1 and "KillMode=process" in warned[0] cron = process_registry.restart_safe_gateway_child_argv( ["hermes", "cron"], unit_suffix="cron-job-1", require_restart_safe_scope=True, ) assert cron.mode == "in_process" @pytest.mark.platforms("linux") def test_real_user_systemd_scope_preserves_worker_context( request, worker_setup: tuple[Path, kb.Task], monkeypatch: pytest.MonkeyPatch ) -> None: from tools import process_registry if not process_registry._systemd_run_user_scope_available(): pytest.skip("systemd-run --user --scope is unavailable on this host") workspace, task = worker_setup from dataclasses import replace from uuid import uuid4 # User-manager transient units outlive pytest's temporary directories. task = replace(task, id=f"t_candidate_{uuid4().hex}") unit = f"hermes-worker-kanban-{task.id}-run-{task.current_run_id}.scope" # Only this freshly named scope is test-owned; the live-system guard still refuses every other unit. request.node.live_system_guard_owned_units = {unit} receipt = workspace / "worker-receipt.json" script = ( "import json, os, pathlib, sys, time; " "p = pathlib.Path(sys.argv[1]); t = p.with_suffix('.tmp'); " "t.write_text(json.dumps({" "'pid': os.getpid(), 'cwd': os.getcwd(), " "'task': os.environ.get('HERMES_KANBAN_TASK'), " "'run': os.environ.get('HERMES_KANBAN_RUN_ID'), " "'claim': os.environ.get('HERMES_KANBAN_CLAIM_LOCK'), " "'board': os.environ.get('HERMES_KANBAN_BOARD'), " "'home': os.environ.get('HERMES_HOME'), " "'profile': os.environ.get('HERMES_PROFILE'), " "'source': os.environ.get('HERMES_SESSION_SOURCE'), " "'terminal_cwd': os.environ.get('TERMINAL_CWD'), " "'cgroup': pathlib.Path('/proc/self/cgroup').read_text()})); os.replace(t, p); time.sleep(0.5)" ) # Replace only the worker payload; keep real scope creation and env handoff. monkeypatch.setattr( kbd, "_worker_argv", lambda task, profile_arg, hermes_home: [sys.executable, "-c", script, str(receipt)], ) monkeypatch.setenv("INVOCATION_ID", "managed-gateway-test") monkeypatch.setattr(process_registry, "_is_supervised_gateway_process", lambda: True) pid = kbd._default_spawn(task, str(workspace)) try: deadline = time.monotonic() + 15 while not receipt.exists() and time.monotonic() < deadline: time.sleep(0.05) log_path = kb.worker_logs_dir() / f"{task.id}.log" assert receipt.exists(), log_path.read_text(encoding="utf-8", errors="replace") payload = json.loads(receipt.read_text(encoding="utf-8")) assert payload["pid"] == pid assert payload["cwd"] == str(workspace) assert payload["task"] == task.id assert payload["run"] == "23" assert payload["claim"] == task.claim_lock assert payload["board"] == kb.get_current_board() assert payload["home"] == str(workspace.parent / '.hermes' / 'profiles' / task.assignee) assert payload["profile"] == task.assignee assert payload["source"] == 'kanban' assert payload["terminal_cwd"] == str(workspace) assert unit in payload["cgroup"] assert "hermes-gateway.service" not in payload["cgroup"] import psutil assert psutil.Process(pid).wait(timeout=5) == 0 finally: # Stop only this test's freshly named scope, never the historical fixed # name or another producer's unit. The live-system guard stays enabled. subprocess.run(["systemctl", "--user", "stop", unit], stdin=subprocess.DEVNULL, capture_output=True, timeout=10, check=False, env=process_registry.systemd_user_bus_env()) state = subprocess.run(["systemctl", "--user", "show", unit, "--property=ActiveState", "--value"], stdin=subprocess.DEVNULL, capture_output=True, text=True, timeout=10, env=process_registry.systemd_user_bus_env(), check=False) assert state.stdout.strip() in ('inactive', 'failed'), state.stdout + state.stderr print(json.dumps({'worker': payload, 'scope_after_cleanup': state.stdout.strip()}))