diff --git a/hermes_cli/setup.py b/hermes_cli/setup.py index 95c9fa622..72b8aab18 100644 --- a/hermes_cli/setup.py +++ b/hermes_cli/setup.py @@ -2572,9 +2572,120 @@ _OPENCLAW_SCRIPT = ( ) +def _load_openclaw_migration_module(): + """Load the openclaw_to_hermes migration script as a module. + + Returns the loaded module, or None if the script can't be loaded. + """ + if not _OPENCLAW_SCRIPT.exists(): + return None + + spec = importlib.util.spec_from_file_location( + "openclaw_to_hermes", _OPENCLAW_SCRIPT + ) + if spec is None or spec.loader is None: + return None + + mod = importlib.util.module_from_spec(spec) + # Register in sys.modules so @dataclass can resolve the module + # (Python 3.11+ requires this for dynamically loaded modules) + import sys as _sys + _sys.modules[spec.name] = mod + try: + spec.loader.exec_module(mod) + except Exception: + _sys.modules.pop(spec.name, None) + raise + return mod + + +# Item kinds that represent high-impact changes warranting explicit warnings. +# Gateway tokens/channels can hijack messaging platforms from the old agent. +# Config values may have different semantics between OpenClaw and Hermes. +# Instruction/context files (.md) can contain incompatible setup procedures. +_HIGH_IMPACT_KIND_KEYWORDS = { + "gateway": "⚠ Gateway/messaging — this will configure Hermes to use your OpenClaw messaging channels", + "telegram": "⚠ Telegram — this will point Hermes at your OpenClaw Telegram bot", + "slack": "⚠ Slack — this will point Hermes at your OpenClaw Slack workspace", + "discord": "⚠ Discord — this will point Hermes at your OpenClaw Discord bot", + "whatsapp": "⚠ WhatsApp — this will point Hermes at your OpenClaw WhatsApp connection", + "config": "⚠ Config values — OpenClaw settings may not map 1:1 to Hermes equivalents", + "soul": "⚠ Instruction file — may contain OpenClaw-specific setup/restart procedures", + "memory": "⚠ Memory/context file — may reference OpenClaw-specific infrastructure", + "context": "⚠ Context file — may contain OpenClaw-specific instructions", +} + + +def _print_migration_preview(report: dict): + """Print a detailed dry-run preview of what migration would do. + + Groups items by category and adds explicit warnings for high-impact + changes like gateway token takeover and config value differences. + """ + items = report.get("items", []) + if not items: + print_info("Nothing to migrate.") + return + + migrated_items = [i for i in items if i.get("status") == "migrated"] + conflict_items = [i for i in items if i.get("status") == "conflict"] + skipped_items = [i for i in items if i.get("status") == "skipped"] + + warnings_shown = set() + + if migrated_items: + print(color(" Would import:", Colors.GREEN)) + for item in migrated_items: + kind = item.get("kind", "unknown") + dest = item.get("destination", "") + if dest: + dest_short = str(dest).replace(str(Path.home()), "~") + print(f" {kind:<22s} → {dest_short}") + else: + print(f" {kind}") + + # Check for high-impact items and collect warnings + kind_lower = kind.lower() + dest_lower = str(dest).lower() + for keyword, warning in _HIGH_IMPACT_KIND_KEYWORDS.items(): + if keyword in kind_lower or keyword in dest_lower: + warnings_shown.add(warning) + print() + + if conflict_items: + print(color(" Would overwrite (conflicts with existing Hermes config):", Colors.YELLOW)) + for item in conflict_items: + kind = item.get("kind", "unknown") + reason = item.get("reason", "already exists") + print(f" {kind:<22s} {reason}") + print() + + if skipped_items: + print(color(" Would skip:", Colors.DIM)) + for item in skipped_items: + kind = item.get("kind", "unknown") + reason = item.get("reason", "") + print(f" {kind:<22s} {reason}") + print() + + # Print collected warnings + if warnings_shown: + print(color(" ── Warnings ──", Colors.YELLOW)) + for warning in sorted(warnings_shown): + print(color(f" {warning}", Colors.YELLOW)) + print() + print(color(" Note: OpenClaw config values may have different semantics in Hermes.", Colors.YELLOW)) + print(color(" For example, OpenClaw's tool_call_execution: \"auto\" ≠ Hermes's yolo mode.", Colors.YELLOW)) + print(color(" Instruction files (.md) from OpenClaw may contain incompatible procedures.", Colors.YELLOW)) + print() + + def _offer_openclaw_migration(hermes_home: Path) -> bool: """Detect ~/.openclaw and offer to migrate during first-time setup. + Runs a dry-run first to show the user exactly what would be imported, + overwritten, or taken over. Only executes after explicit confirmation. + Returns True if migration ran successfully, False otherwise. """ openclaw_dir = Path.home() / ".openclaw" @@ -2587,12 +2698,12 @@ def _offer_openclaw_migration(hermes_home: Path) -> bool: print() print_header("OpenClaw Installation Detected") print_info(f"Found OpenClaw data at {openclaw_dir}") - print_info("Hermes can import your settings, memories, skills, and API keys.") + print_info("Hermes can preview what would be imported before making any changes.") print() - if not prompt_yes_no("Would you like to import from OpenClaw?", default=True): + if not prompt_yes_no("Would you like to see what can be imported?", default=True): print_info( - "Skipping migration. You can run it later via the openclaw-migration skill." + "Skipping migration. You can run it later with: hermes claw migrate --dry-run" ) return False @@ -2601,34 +2712,71 @@ def _offer_openclaw_migration(hermes_home: Path) -> bool: if not config_path.exists(): save_config(load_config()) - # Dynamically load the migration script + # Load the migration module try: - spec = importlib.util.spec_from_file_location( - "openclaw_to_hermes", _OPENCLAW_SCRIPT - ) - if spec is None or spec.loader is None: + mod = _load_openclaw_migration_module() + if mod is None: print_warning("Could not load migration script.") return False + except Exception as e: + print_warning(f"Could not load migration script: {e}") + logger.debug("OpenClaw migration module load error", exc_info=True) + return False - mod = importlib.util.module_from_spec(spec) - # Register in sys.modules so @dataclass can resolve the module - # (Python 3.11+ requires this for dynamically loaded modules) - import sys as _sys - _sys.modules[spec.name] = mod - try: - spec.loader.exec_module(mod) - except Exception: - _sys.modules.pop(spec.name, None) - raise - - # Run migration with the "full" preset, execute mode, no overwrite + # ── Phase 1: Dry-run preview ── + try: selected = mod.resolve_selected_options(None, None, preset="full") + dry_migrator = mod.Migrator( + source_root=openclaw_dir.resolve(), + target_root=hermes_home.resolve(), + execute=False, # dry-run — no files modified + workspace_target=None, + overwrite=True, # show everything including conflicts + migrate_secrets=True, + output_dir=None, + selected_options=selected, + preset_name="full", + ) + preview_report = dry_migrator.migrate() + except Exception as e: + print_warning(f"Migration preview failed: {e}") + logger.debug("OpenClaw migration preview error", exc_info=True) + return False + + # Display the full preview + preview_summary = preview_report.get("summary", {}) + preview_count = preview_summary.get("migrated", 0) + + if preview_count == 0: + print() + print_info("Nothing to import from OpenClaw.") + return False + + print() + print_header(f"Migration Preview — {preview_count} item(s) would be imported") + print_info("No changes have been made yet. Review the list below:") + print() + _print_migration_preview(preview_report) + + # ── Phase 2: Confirm and execute ── + if not prompt_yes_no("Proceed with migration?", default=False): + print_info( + "Migration cancelled. You can run it later with: hermes claw migrate" + ) + print_info( + "Use --dry-run to preview again, or --preset minimal for a lighter import." + ) + return False + + # Execute the migration — overwrite=False so existing Hermes configs are + # preserved. The user saw the preview; conflicts are skipped by default. + try: migrator = mod.Migrator( source_root=openclaw_dir.resolve(), target_root=hermes_home.resolve(), execute=True, workspace_target=None, - overwrite=True, + overwrite=False, # preserve existing Hermes config migrate_secrets=True, output_dir=None, selected_options=selected, @@ -2640,7 +2788,7 @@ def _offer_openclaw_migration(hermes_home: Path) -> bool: logger.debug("OpenClaw migration error", exc_info=True) return False - # Print summary + # Print final summary summary = report.get("summary", {}) migrated = summary.get("migrated", 0) skipped = summary.get("skipped", 0) @@ -2651,7 +2799,7 @@ def _offer_openclaw_migration(hermes_home: Path) -> bool: if migrated: print_success(f"Imported {migrated} item(s) from OpenClaw.") if conflicts: - print_info(f"Skipped {conflicts} item(s) that already exist in Hermes.") + print_info(f"Skipped {conflicts} item(s) that already exist in Hermes (use hermes claw migrate --overwrite to force).") if skipped: print_info(f"Skipped {skipped} item(s) (not found or unchanged).") if errors: diff --git a/tests/hermes_cli/test_setup_openclaw_migration.py b/tests/hermes_cli/test_setup_openclaw_migration.py index b956f1fe6..fe8026390 100644 --- a/tests/hermes_cli/test_setup_openclaw_migration.py +++ b/tests/hermes_cli/test_setup_openclaw_migration.py @@ -44,7 +44,7 @@ class TestOfferOpenclawMigration: assert setup_mod._offer_openclaw_migration(tmp_path / ".hermes") is False def test_runs_migration_when_user_accepts(self, tmp_path): - """Should dynamically load the script and run the Migrator.""" + """Should run dry-run preview first, then execute after confirmation.""" openclaw_dir = tmp_path / ".openclaw" openclaw_dir.mkdir() @@ -60,6 +60,7 @@ class TestOfferOpenclawMigration: fake_migrator = MagicMock() fake_migrator.migrate.return_value = { "summary": {"migrated": 3, "skipped": 1, "conflict": 0, "error": 0}, + "items": [{"kind": "config", "status": "migrated", "destination": "/tmp/x"}], "output_dir": str(hermes_home / "migration"), } fake_mod.Migrator = MagicMock(return_value=fake_migrator) @@ -70,6 +71,7 @@ class TestOfferOpenclawMigration: with ( patch("hermes_cli.setup.Path.home", return_value=tmp_path), patch.object(setup_mod, "_OPENCLAW_SCRIPT", script), + # Both prompts answered Yes: preview offer + proceed confirmation patch.object(setup_mod, "prompt_yes_no", return_value=True), patch.object(setup_mod, "get_config_path", return_value=config_path), patch("importlib.util.spec_from_file_location") as mock_spec_fn, @@ -91,13 +93,75 @@ class TestOfferOpenclawMigration: fake_mod.resolve_selected_options.assert_called_once_with( None, None, preset="full" ) - fake_mod.Migrator.assert_called_once() - call_kwargs = fake_mod.Migrator.call_args[1] - assert call_kwargs["execute"] is True - assert call_kwargs["overwrite"] is True - assert call_kwargs["migrate_secrets"] is True - assert call_kwargs["preset_name"] == "full" - fake_migrator.migrate.assert_called_once() + # Migrator called twice: once for dry-run preview, once for execution + assert fake_mod.Migrator.call_count == 2 + + # First call: dry-run preview (execute=False, overwrite=True to show all) + preview_kwargs = fake_mod.Migrator.call_args_list[0][1] + assert preview_kwargs["execute"] is False + assert preview_kwargs["overwrite"] is True + assert preview_kwargs["migrate_secrets"] is True + assert preview_kwargs["preset_name"] == "full" + + # Second call: actual execution (execute=True, overwrite=False to preserve) + exec_kwargs = fake_mod.Migrator.call_args_list[1][1] + assert exec_kwargs["execute"] is True + assert exec_kwargs["overwrite"] is False + assert exec_kwargs["migrate_secrets"] is True + assert exec_kwargs["preset_name"] == "full" + + # migrate() called twice (once per Migrator instance) + assert fake_migrator.migrate.call_count == 2 + + def test_user_declines_after_preview(self, tmp_path): + """Should return False when user sees preview but declines to proceed.""" + openclaw_dir = tmp_path / ".openclaw" + openclaw_dir.mkdir() + + hermes_home = tmp_path / ".hermes" + hermes_home.mkdir() + config_path = hermes_home / "config.yaml" + config_path.write_text("agent:\n max_turns: 90\n") + + fake_mod = ModuleType("openclaw_to_hermes") + fake_mod.resolve_selected_options = MagicMock(return_value={"soul", "memory"}) + fake_migrator = MagicMock() + fake_migrator.migrate.return_value = { + "summary": {"migrated": 3, "skipped": 0, "conflict": 0, "error": 0}, + "items": [{"kind": "config", "status": "migrated", "destination": "/tmp/x"}], + } + fake_mod.Migrator = MagicMock(return_value=fake_migrator) + + script = tmp_path / "openclaw_to_hermes.py" + script.write_text("# placeholder") + + # First prompt (preview): Yes, Second prompt (proceed): No + prompt_responses = iter([True, False]) + + with ( + patch("hermes_cli.setup.Path.home", return_value=tmp_path), + patch.object(setup_mod, "_OPENCLAW_SCRIPT", script), + patch.object(setup_mod, "prompt_yes_no", side_effect=prompt_responses), + patch.object(setup_mod, "get_config_path", return_value=config_path), + patch("importlib.util.spec_from_file_location") as mock_spec_fn, + ): + mock_spec = MagicMock() + mock_spec.loader = MagicMock() + mock_spec_fn.return_value = mock_spec + + def exec_module(mod): + mod.resolve_selected_options = fake_mod.resolve_selected_options + mod.Migrator = fake_mod.Migrator + + mock_spec.loader.exec_module = exec_module + + result = setup_mod._offer_openclaw_migration(hermes_home) + + assert result is False + # Only dry-run Migrator was created, not the execute one + assert fake_mod.Migrator.call_count == 1 + preview_kwargs = fake_mod.Migrator.call_args[1] + assert preview_kwargs["execute"] is False def test_handles_migration_error_gracefully(self, tmp_path): """Should catch exceptions and return False."""