fix(claw): warn if OpenClaw is running before migration

agent/context_compressor.py

@@ -78,6 +78,7 @@ def on_session_reset(self) -> None:
         self._context_probed = False
         self._context_probe_persistable = False
         self._previous_summary = None
+        self._summary_failure_cooldown_until = 0.0
 
     def update_model(
         self,

hermes_cli/claw.py

@@ -11,6 +11,7 @@
 
 import importlib.util
 import logging
+import subprocess
 import sys
 from datetime import datetime
 from pathlib import Path
@@ -52,6 +53,73 @@
 # Known OpenClaw directory names (current + legacy)
 _OPENCLAW_DIR_NAMES = (".openclaw", ".clawdbot", ".moldbot")
 
+def _is_openclaw_running() -> bool:
+    """Check whether an OpenClaw process appears to be running."""
+    if sys.platform == "win32":
+        try:
+            # First check for dedicated executables
+            for exe in ("openclaw.exe", "clawd.exe"):
+                result = subprocess.run(
+                    ["tasklist", "/FI", f"IMAGENAME eq {exe}"],
+                    capture_output=True, text=True, timeout=5
+                )
+                if exe in result.stdout.lower():
+                    return True
+
+            # Check node.exe processes for openclaw/clawd in command line.
+            # tasklist does not include command lines, so we use PowerShell.
+            ps_cmd = (
+                'Get-CimInstance Win32_Process -Filter "Name = \'node.exe\'" | '
+                'Where-Object { $_.CommandLine -match "openclaw|clawd" } | '
+                'Select-Object -First 1 ProcessId'
+            )
+            result = subprocess.run(
+                ["powershell", "-NoProfile", "-Command", ps_cmd],
+                capture_output=True, text=True, timeout=5
+            )
+            return bool(result.stdout.strip())
+        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 auto_yes:
+        return
+    if not sys.stdin.isatty():
+        print_info("Non-interactive session — continuing to preview only.")
+        return
+    if 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 +355,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

tests/hermes_cli/test_claw.py

@@ -197,6 +197,11 @@ def test_shows_help_for_no_action(self, capsys):
 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"),
@@ -730,3 +735,109 @@ def test_empty_report(self, capsys):
         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_when_openclaw_exe_running(self):
+        with patch.object(claw_mod, "sys") as mock_sys:
+            mock_sys.platform = "win32"
+            with patch.object(claw_mod, "subprocess") as mock_subprocess:
+                # First tasklist (openclaw.exe) matches
+                mock_subprocess.run.side_effect = [
+                    MagicMock(returncode=0, stdout="openclaw.exe                 1234 Console    1     45,056 K\n"),
+                ]
+                assert claw_mod._is_openclaw_running() is True
+
+    def test_returns_true_on_windows_when_node_exe_has_openclaw_in_cmdline(self):
+        with patch.object(claw_mod, "sys") as mock_sys:
+            mock_sys.platform = "win32"
+            with patch.object(claw_mod, "subprocess") as mock_subprocess:
+                # tasklist for openclaw.exe and clawd.exe both miss,
+                # PowerShell finds a matching node.exe process.
+                mock_subprocess.run.side_effect = [
+                    MagicMock(returncode=0, stdout=""),
+                    MagicMock(returncode=0, stdout=""),
+                    MagicMock(returncode=0, stdout="1234\n"),
+                ]
+                assert claw_mod._is_openclaw_running() is True
+
+    def test_returns_false_on_windows_when_node_exe_has_no_openclaw_in_cmdline(self):
+        with patch.object(claw_mod, "sys") as mock_sys:
+            mock_sys.platform = "win32"
+            with patch.object(claw_mod, "subprocess") as mock_subprocess:
+                # Neither dedicated exe nor PowerShell find anything.
+                mock_subprocess.run.side_effect = [
+                    MagicMock(returncode=0, stdout=""),
+                    MagicMock(returncode=0, stdout=""),
+                    MagicMock(returncode=0, stdout=""),
+                ]
+                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 patch.object(claw_mod.sys.stdin, "isatty", return_value=True):
+                    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):
+                with patch.object(claw_mod.sys.stdin, "isatty", 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
+
+    def test_warns_and_continues_in_non_interactive_session(self, capsys):
+        with patch.object(claw_mod, "_is_openclaw_running", return_value=True):
+            with patch.object(claw_mod.sys.stdin, "isatty", return_value=False):
+                claw_mod._warn_if_openclaw_running(auto_yes=False)
+        captured = capsys.readouterr()
+        assert "OpenClaw appears to be running" in captured.out
+        assert "Non-interactive session" in captured.out