From 5af9614f6d3df91ca9f2a6e45b2d43f6e6adde67 Mon Sep 17 00:00:00 2001 From: dirtyfancy Date: Sun, 12 Apr 2026 10:34:38 +0800 Subject: [PATCH] fix(claw): warn if OpenClaw is running before migration Add _is_openclaw_running() and _warn_if_openclaw_running() to detect OpenClaw processes (via pgrep/tasklist) before hermes claw migrate. Warns the user that messaging platforms only allow one active session per bot token, and lets them cancel or continue. Fixes #7907 --- hermes_cli/claw.py | 55 +++++++++++++++++++++- tests/hermes_cli/test_claw.py | 86 +++++++++++++++++++++++++++++++++++ 2 files changed, 139 insertions(+), 2 deletions(-) diff --git a/hermes_cli/claw.py b/hermes_cli/claw.py index 0f9e28cbcc..bc38ee7a66 100644 --- a/hermes_cli/claw.py +++ b/hermes_cli/claw.py @@ -11,6 +11,7 @@ Usage: import importlib.util import logging +import subprocess import sys from datetime import datetime from pathlib import Path @@ -52,6 +53,53 @@ _OPENCLAW_SCRIPT_INSTALLED = ( # Known OpenClaw directory names (current + legacy) _OPENCLAW_DIR_NAMES = (".openclaw", ".clawdbot", ".moltbot") +def _is_openclaw_running() -> bool: + """Check whether an OpenClaw process appears to be running.""" + if sys.platform == "win32": + try: + result = subprocess.run( + ["tasklist", "/FI", "IMAGENAME eq node.exe"], + capture_output=True, text=True, timeout=5 + ) + output = result.stdout.lower() + return "openclaw" in output or "clawd" in output + except Exception: + return False + + for cmd in (["pgrep", "-f", "openclaw"], ["pgrep", "-f", "clawd"]): + try: + result = subprocess.run(cmd, capture_output=True, timeout=3) + if result.returncode == 0: + return True + except (FileNotFoundError, subprocess.TimeoutExpired): + continue + return False + + +def _warn_if_openclaw_running(auto_yes: bool) -> None: + """Warn if OpenClaw is still running before migration. + + Telegram, Discord, and Slack only allow one active connection per bot + token. Migrating while OpenClaw is running causes both to fight for the + same token. + """ + if not _is_openclaw_running(): + return + + print() + print_error("OpenClaw appears to be running.") + print_info( + "Messaging platforms (Telegram, Discord, Slack) only allow one " + "active session per bot token. If you continue, both OpenClaw and " + "Hermes may try to use the same token, causing disconnects." + ) + print_info("Recommendation: stop OpenClaw before migrating.") + print() + if not auto_yes and not prompt_yes_no("Continue anyway?", default=False): + print_info("Migration cancelled. Stop OpenClaw and try again.") + sys.exit(0) + + def _warn_if_gateway_running(auto_yes: bool) -> None: """Check if a Hermes gateway is running with connected platforms. @@ -287,8 +335,11 @@ def _cmd_migrate(args): print_info(f"Workspace: {workspace_target}") print() - # Check if a gateway is running with connected platforms — migrating tokens - # while the gateway is active will cause conflicts (e.g. Telegram 409). + # Check if OpenClaw is still running — migrating tokens while both are + # active will cause conflicts (e.g. Telegram 409). + _warn_if_openclaw_running(auto_yes) + + # Check if a Hermes gateway is running with connected platforms. _warn_if_gateway_running(auto_yes) # Ensure config.yaml exists before migration tries to read it diff --git a/tests/hermes_cli/test_claw.py b/tests/hermes_cli/test_claw.py index d7528890e2..dc9024d6bc 100644 --- a/tests/hermes_cli/test_claw.py +++ b/tests/hermes_cli/test_claw.py @@ -197,6 +197,11 @@ class TestClawCommand: class TestCmdMigrate: """Test the migrate command handler.""" + @pytest.fixture(autouse=True) + def _mock_openclaw_running(self): + with patch.object(claw_mod, "_is_openclaw_running", return_value=False): + yield + def test_error_when_source_missing(self, tmp_path, capsys): args = Namespace( source=str(tmp_path / "nonexistent"), @@ -626,3 +631,84 @@ class TestPrintMigrationReport: claw_mod._print_migration_report(report, dry_run=False) captured = capsys.readouterr() assert "Nothing to migrate" in captured.out + + +class TestIsOpenclawRunning: + def test_returns_true_when_pgrep_finds_openclaw(self): + with patch.object(claw_mod, "sys") as mock_sys: + mock_sys.platform = "darwin" + with patch.object(claw_mod, "subprocess") as mock_subprocess: + mock_subprocess.run.side_effect = [ + MagicMock(returncode=0), + ] + assert claw_mod._is_openclaw_running() is True + + def test_returns_true_when_pgrep_finds_clawd(self): + with patch.object(claw_mod, "sys") as mock_sys: + mock_sys.platform = "linux" + with patch.object(claw_mod, "subprocess") as mock_subprocess: + mock_subprocess.run.side_effect = [ + MagicMock(returncode=1), + MagicMock(returncode=0), + ] + assert claw_mod._is_openclaw_running() is True + + def test_returns_false_when_pgrep_finds_nothing(self): + with patch.object(claw_mod, "sys") as mock_sys: + mock_sys.platform = "darwin" + with patch.object(claw_mod, "subprocess") as mock_subprocess: + mock_subprocess.run.side_effect = [ + MagicMock(returncode=1), + MagicMock(returncode=1), + ] + assert claw_mod._is_openclaw_running() is False + + def test_returns_true_on_windows_tasklist(self): + with patch.object(claw_mod, "sys") as mock_sys: + mock_sys.platform = "win32" + with patch.object(claw_mod, "subprocess") as mock_subprocess: + mock_subprocess.run.return_value = MagicMock( + returncode=0, + stdout="node.exe openclaw-gateway", + ) + assert claw_mod._is_openclaw_running() is True + + def test_returns_false_on_windows_when_not_found(self): + with patch.object(claw_mod, "sys") as mock_sys: + mock_sys.platform = "win32" + with patch.object(claw_mod, "subprocess") as mock_subprocess: + mock_subprocess.run.return_value = MagicMock( + returncode=0, + stdout="node.exe some-other-app", + ) + assert claw_mod._is_openclaw_running() is False + + +class TestWarnIfOpenclawRunning: + def test_noop_when_not_running(self, capsys): + with patch.object(claw_mod, "_is_openclaw_running", return_value=False): + claw_mod._warn_if_openclaw_running(auto_yes=False) + captured = capsys.readouterr() + assert captured.out == "" + + def test_warns_and_exits_when_running_and_user_declines(self, capsys): + with patch.object(claw_mod, "_is_openclaw_running", return_value=True): + with patch.object(claw_mod, "prompt_yes_no", return_value=False): + with pytest.raises(SystemExit) as exc_info: + claw_mod._warn_if_openclaw_running(auto_yes=False) + assert exc_info.value.code == 0 + captured = capsys.readouterr() + assert "OpenClaw appears to be running" in captured.out + + def test_warns_and_continues_when_running_and_user_accepts(self, capsys): + with patch.object(claw_mod, "_is_openclaw_running", return_value=True): + with patch.object(claw_mod, "prompt_yes_no", return_value=True): + claw_mod._warn_if_openclaw_running(auto_yes=False) + captured = capsys.readouterr() + assert "OpenClaw appears to be running" in captured.out + + def test_warns_and_continues_in_auto_yes_mode(self, capsys): + with patch.object(claw_mod, "_is_openclaw_running", return_value=True): + claw_mod._warn_if_openclaw_running(auto_yes=True) + captured = capsys.readouterr() + assert "OpenClaw appears to be running" in captured.out