Telegram gateway stops on headless server logout

Fixes on record

Merged to main. A stamp counts the issues a fix is linked to.

• #1334 fix: auto-enable systemd linger during gateway install on headless servers

  merged to main

  files: hermes_cli/gateway.py, tests/hermes_cli/test_gateway.py, tests/hermes_cli/test_gateway_linger.py

  Show the diff

  • hermes_cli/gateway.py

    @@ -251,7 +251,6 @@ def generate_systemd_unit() -> str:
     WantedBy=default.target
     """
     
    -
     def _normalize_service_definition(text: str) -> str:
         return "\n".join(line.rstrip() for line in text.strip().splitlines())
     
    @@ -279,6 +278,65 @@ def refresh_systemd_unit_if_needed() -> bool:
         return True
     
     
    +
    +def _print_linger_enable_warning(username: str, detail: str | None = None) -> None:
    +    print()
    +    print("⚠ Linger not enabled — gateway may stop when you close this terminal.")
    +    if detail:
    +        print(f"  Auto-enable failed: {detail}")
    +    print()
    +    print("  On headless servers (VPS, cloud instances) run:")
    +    print(f"    sudo loginctl enable-linger {username}")
    +    print()
    +    print("  Then restart the gateway:")
    +    print(f"    systemctl --user restart {SERVICE_NAME}.service")
    +    print()
    +
    +
    +
    +def _ensure_linger_enabled() -> None:
    +    """Enable linger when possible so the user gateway survives logout."""
    +    if not is_linux():
    +        return
    +
    +    import getpass
    +    import shutil
    +
    +    username = getpass.getuser()
    +    linger_file = Path(f"/var/lib/systemd/linger/{username}")
    +    if linger_file.exists():
    +        print("✓ Systemd linger is enabled (service survives logout)")
    +        return
    +
    +    linger_enabled, linger_detail = get_systemd_linger_status()
    +    if linger_enabled is True:
    +        print("✓ Systemd linger is enabled (service survives logout)")
    +        return
    +
    +    if not shutil.which("loginctl"):
    +        _print_linger_enable_warning(username, linger_detail or "loginctl not found")
    +        return
    +
    +    print("Enabling linger so the gateway survives SSH logout...")
    +    try:
    +        result = subprocess.run(
    +            ["loginctl", "enable-linger", username],
    +            capture_output=True,
    +            text=True,
    +            check=False,
    +        )
    +    except Exception as e:
    +        _print_linger_enable_warning(username, str(e))
    +        return
    +
    +    if result.returncode == 0:
    +        print("✓ Linger enabled — gateway will persist after logout")
    +        return
    +
    +    detail = (result.stderr or result.stdout or f"exit {result.returncode}").strip()
    +    _print_linger_enable_warning(username, detail or linger_detail)
    +
    +
     def systemd_install(force: bool = False):
         unit_path = get_systemd_unit_path()
         
    @@ -302,7 +360,7 @@ def systemd_install(force: bool = False):
         print(f"  hermes gateway status             # Check status")
         print(f"  journalctl --user -u {SERVICE_NAME} -f  # View logs")
         print()
    -    print_systemd_linger_guidance()
    +    _ensure_linger_enabled()
     
     def systemd_uninstall():
         subprocess.run(["systemctl", "--user", "stop", SERVICE_NAME], check=False)

  • tests/hermes_cli/test_gateway.py

    @@ -59,15 +59,16 @@ def test_systemd_install_checks_linger_status(monkeypatch, tmp_path, capsys):
         unit_path = tmp_path / "systemd" / "user" / "hermes-gateway.service"
     
         monkeypatch.setattr(gateway, "get_systemd_unit_path", lambda: unit_path)
    -    monkeypatch.setattr(gateway, "get_systemd_linger_status", lambda: (False, ""))
     
         calls = []
    +    helper_calls = []
     
         def fake_run(cmd, check=False, **kwargs):
             calls.append((cmd, check))
             return SimpleNamespace(returncode=0, stdout="", stderr="")
     
         monkeypatch.setattr(gateway.subprocess, "run", fake_run)
    +    monkeypatch.setattr(gateway, "_ensure_linger_enabled", lambda: helper_calls.append(True))
     
         gateway.systemd_install(force=False)
     
    @@ -77,6 +78,5 @@ def fake_run(cmd, check=False, **kwargs):
             ["systemctl", "--user", "daemon-reload"],
             ["systemctl", "--user", "enable", gateway.SERVICE_NAME],
         ]
    +    assert helper_calls == [True]
         assert "Service installed and enabled" in out
    -    assert "Systemd linger is disabled" in out
    -    assert "loginctl enable-linger" in out

  • tests/hermes_cli/test_gateway_linger.py

    @@ -0,0 +1,120 @@
    +"""Tests for gateway linger auto-enable behavior on headless Linux installs."""
    +
    +from types import SimpleNamespace
    +
    +import hermes_cli.gateway as gateway
    +
    +
    +class TestEnsureLingerEnabled:
    +    def test_linger_already_enabled_via_file(self, monkeypatch, capsys):
    +        monkeypatch.setattr(gateway, "is_linux", lambda: True)
    +        monkeypatch.setattr("getpass.getuser", lambda: "testuser")
    +        monkeypatch.setattr(gateway, "Path", lambda _path: SimpleNamespace(exists=lambda: True))
    +
    +        calls = []
    +        monkeypatch.setattr(gateway.subprocess, "run", lambda *args, **kwargs: calls.append((args, kwargs)))
    +
    +        gateway._ensure_linger_enabled()
    +
    +        out = capsys.readouterr().out
    +        assert "Systemd linger is enabled" in out
    +        assert calls == []
    +
    +    def test_status_enabled_skips_enable(self, monkeypatch, capsys):
    +        monkeypatch.setattr(gateway, "is_linux", lambda: True)
    +        monkeypatch.setattr("getpass.getuser", lambda: "testuser")
    +        monkeypatch.setattr(gateway, "Path", lambda _path: SimpleNamespace(exists=lambda: False))
    +        monkeypatch.setattr(gateway, "get_systemd_linger_status", lambda: (True, ""))
    +
    +        calls = []
    +        monkeypatch.setattr(gateway.subprocess, "run", lambda *args, **kwargs: calls.append((args, kwargs)))
    +
    +        gateway._ensure_linger_enabled()
    +
    +        out = capsys.readouterr().out
    +        assert "Systemd linger is enabled" in out
    +        assert calls == []
    +
    +    def test_loginctl_success_enables_linger(self, monkeypatch, capsys):
    +        monkeypatch.setattr(gateway, "is_linux", lambda: True)
    +        monkeypatch.setattr("getpass.getuser", lambda: "testuser")
    +        monkeypatch.setattr(gateway, "Path", lambda _path: SimpleNamespace(exists=lambda: False))
    +        monkeypatch.setattr(gateway, "get_systemd_linger_status", lambda: (False, ""))
    +        monkeypatch.setattr("shutil.which", lambda name: "/usr/bin/loginctl")
    +
    +        run_calls = []
    +
    +        def fake_run(cmd, capture_output=False, text=False, check=False):
    +            run_calls.append((cmd, capture_output, text, check))
    +            return SimpleNamespace(returncode=0, stdout="", stderr="")
    +
    +        monkeypatch.setattr(gateway.subprocess, "run", fake_run)
    +
    +        gateway._ensure_linger_enabled()
    +
    +        out = capsys.readouterr().out
    +        assert "Enabling linger" in out
    +        assert "Linger enabled" in out
    +        assert run_calls == [(["loginctl", "enable-linger", "testuser"], True, True, False)]
    +
    +    def test_missing_loginctl_shows_manual_guidance(self, monkeypatch, capsys):
    +        monkeypatch.setattr(gateway, "is_linux", lambda: True)
    +        monkeypatch.setattr("getpass.getuser", lambda: "testuser")
    +        monkeypatch.setattr(gateway, "Path", lambda _path: SimpleNamespace(exists=lambda: False))
    +        monkeypatch.setattr(gateway, "get_systemd_linger_status", lambda: (None, "loginctl not found"))
    +        monkeypatch.setattr("shutil.which", lambda name: None)
    +
    +        calls = []
    +        monkeypatch.setattr(gateway.subprocess, "run", lambda *args, **kwargs: calls.append((args, kwargs)))
    +
    +        gateway._ensure_linger_enabled()
    +
    +        out = capsys.readouterr().out
    +        assert "sudo loginctl enable-linger testuser" in out
    +        assert "loginctl not found" in out
    +        assert calls == []
    +
    +    def test_loginctl_failure_shows_manual_guidance(self, monkeypatch, capsys):
    +        monkeypatch.setattr(gateway, "is_linux", lambda: True)
    +        monkeypatch.setattr("getpass.getuser", lambda: "testuser")
    +        monkeypatch.setattr(gateway, "Path", lambda _path: SimpleNamespace(exists=lambda: False))
    +        monkeypatch.setattr(gateway, "get_systemd_linger_status", lambda: (False, ""))
    +        monkeypatch.setattr("shutil.which", lambda name: "/usr/bin/loginctl")
    +        monkeypatch.setattr(
    +            gateway.subprocess,
    +            "run",
    +            lambda *args, **kwargs: SimpleNamespace(returncode=1, stdout="", stderr="Permission denied"),
    +        )
    +
    +        gateway._ensure_linger_enabled()
    +
    +        out = capsys.readouterr().out
    +        assert "sudo loginctl enable-linger testuser" in out
    +        assert "Permission denied" in out
    +
    +
    +def test_systemd_install_calls_linger_helper(monkeypatch, tmp_path, capsys):
    +    unit_path = tmp_path / "systemd" / "user" / "hermes-gateway.service"
    +
    +    monkeypatch.setattr(gateway, "get_systemd_unit_path", lambda: unit_path)
    +
    +    calls = []
    +
    +    def fake_run(cmd, check=False, **kwargs):
    +        calls.append((cmd, check))
    +        return SimpleNamespace(returncode=0, stdout="", stderr="")
    +
    +    helper_calls = []
    +    monkeypatch.setattr(gateway.subprocess, "run", fake_run)
    +    monkeypatch.setattr(gateway, "_ensure_linger_enabled", lambda: helper_calls.append(True))
    +
    +    gateway.systemd_install(force=False)
    +
    +    out = capsys.readouterr().out
    +    assert unit_path.exists()
    +    assert [cmd for cmd, _ in calls] == [
    +        ["systemctl", "--user", "daemon-reload"],
    +        ["systemctl", "--user", "enable", gateway.SERVICE_NAME],
    +    ]
    +    assert helper_calls == [True]
    +    assert "Service installed and enabled" in out