Compare commits

...

1 Commits

Author SHA1 Message Date
Teknium
6e8a6203cd fix: OpenClaw migration now shows dry-run preview before executing
The setup wizard's OpenClaw migration previously ran immediately with
aggressive defaults (overwrite=True, preset=full) after a single
'Would you like to import?' prompt. This caused several problems:

- Config values with different semantics (e.g. tool_call_execution:
  'auto' in OpenClaw vs 'off' for Hermes yolo mode) were imported
  without translation
- Gateway tokens were hijacked from OpenClaw without warning, taking
  over Telegram/Slack/Discord channels
- Instruction files (.md) containing OpenClaw-specific setup/restart
  procedures were copied, causing Hermes restart failures

Now the migration:
1. Asks 'Would you like to see what can be imported?' (softer framing)
2. Runs a dry-run preview showing everything that would be imported
3. Displays categorized warnings for high-impact items (gateway
   takeover, config value differences, instruction files)
4. Asks for explicit confirmation with default=No
5. Executes with overwrite=False (preserves existing Hermes config)

Also extracts _load_openclaw_migration_module() for reuse and adds
_print_migration_preview() with keyword-based warning detection.

Tests updated for two-phase behavior + new test for decline-after-preview.
2026-04-09 11:31:51 -07:00
2 changed files with 243 additions and 31 deletions

View File

@@ -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: def _offer_openclaw_migration(hermes_home: Path) -> bool:
"""Detect ~/.openclaw and offer to migrate during first-time setup. """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. Returns True if migration ran successfully, False otherwise.
""" """
openclaw_dir = Path.home() / ".openclaw" openclaw_dir = Path.home() / ".openclaw"
@@ -2587,12 +2698,12 @@ def _offer_openclaw_migration(hermes_home: Path) -> bool:
print() print()
print_header("OpenClaw Installation Detected") print_header("OpenClaw Installation Detected")
print_info(f"Found OpenClaw data at {openclaw_dir}") 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() 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( 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 return False
@@ -2601,34 +2712,71 @@ def _offer_openclaw_migration(hermes_home: Path) -> bool:
if not config_path.exists(): if not config_path.exists():
save_config(load_config()) save_config(load_config())
# Dynamically load the migration script # Load the migration module
try: try:
spec = importlib.util.spec_from_file_location( mod = _load_openclaw_migration_module()
"openclaw_to_hermes", _OPENCLAW_SCRIPT if mod is None:
)
if spec is None or spec.loader is None:
print_warning("Could not load migration script.") print_warning("Could not load migration script.")
return False 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) # ── Phase 1: Dry-run preview ──
# Register in sys.modules so @dataclass can resolve the module try:
# (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
selected = mod.resolve_selected_options(None, None, preset="full") 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( migrator = mod.Migrator(
source_root=openclaw_dir.resolve(), source_root=openclaw_dir.resolve(),
target_root=hermes_home.resolve(), target_root=hermes_home.resolve(),
execute=True, execute=True,
workspace_target=None, workspace_target=None,
overwrite=True, overwrite=False, # preserve existing Hermes config
migrate_secrets=True, migrate_secrets=True,
output_dir=None, output_dir=None,
selected_options=selected, selected_options=selected,
@@ -2640,7 +2788,7 @@ def _offer_openclaw_migration(hermes_home: Path) -> bool:
logger.debug("OpenClaw migration error", exc_info=True) logger.debug("OpenClaw migration error", exc_info=True)
return False return False
# Print summary # Print final summary
summary = report.get("summary", {}) summary = report.get("summary", {})
migrated = summary.get("migrated", 0) migrated = summary.get("migrated", 0)
skipped = summary.get("skipped", 0) skipped = summary.get("skipped", 0)
@@ -2651,7 +2799,7 @@ def _offer_openclaw_migration(hermes_home: Path) -> bool:
if migrated: if migrated:
print_success(f"Imported {migrated} item(s) from OpenClaw.") print_success(f"Imported {migrated} item(s) from OpenClaw.")
if conflicts: 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: if skipped:
print_info(f"Skipped {skipped} item(s) (not found or unchanged).") print_info(f"Skipped {skipped} item(s) (not found or unchanged).")
if errors: if errors:

View File

@@ -44,7 +44,7 @@ class TestOfferOpenclawMigration:
assert setup_mod._offer_openclaw_migration(tmp_path / ".hermes") is False assert setup_mod._offer_openclaw_migration(tmp_path / ".hermes") is False
def test_runs_migration_when_user_accepts(self, tmp_path): 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 = tmp_path / ".openclaw"
openclaw_dir.mkdir() openclaw_dir.mkdir()
@@ -60,6 +60,7 @@ class TestOfferOpenclawMigration:
fake_migrator = MagicMock() fake_migrator = MagicMock()
fake_migrator.migrate.return_value = { fake_migrator.migrate.return_value = {
"summary": {"migrated": 3, "skipped": 1, "conflict": 0, "error": 0}, "summary": {"migrated": 3, "skipped": 1, "conflict": 0, "error": 0},
"items": [{"kind": "config", "status": "migrated", "destination": "/tmp/x"}],
"output_dir": str(hermes_home / "migration"), "output_dir": str(hermes_home / "migration"),
} }
fake_mod.Migrator = MagicMock(return_value=fake_migrator) fake_mod.Migrator = MagicMock(return_value=fake_migrator)
@@ -70,6 +71,7 @@ class TestOfferOpenclawMigration:
with ( with (
patch("hermes_cli.setup.Path.home", return_value=tmp_path), patch("hermes_cli.setup.Path.home", return_value=tmp_path),
patch.object(setup_mod, "_OPENCLAW_SCRIPT", script), 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, "prompt_yes_no", return_value=True),
patch.object(setup_mod, "get_config_path", return_value=config_path), patch.object(setup_mod, "get_config_path", return_value=config_path),
patch("importlib.util.spec_from_file_location") as mock_spec_fn, 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( fake_mod.resolve_selected_options.assert_called_once_with(
None, None, preset="full" None, None, preset="full"
) )
fake_mod.Migrator.assert_called_once() # Migrator called twice: once for dry-run preview, once for execution
call_kwargs = fake_mod.Migrator.call_args[1] assert fake_mod.Migrator.call_count == 2
assert call_kwargs["execute"] is True
assert call_kwargs["overwrite"] is True # First call: dry-run preview (execute=False, overwrite=True to show all)
assert call_kwargs["migrate_secrets"] is True preview_kwargs = fake_mod.Migrator.call_args_list[0][1]
assert call_kwargs["preset_name"] == "full" assert preview_kwargs["execute"] is False
fake_migrator.migrate.assert_called_once() 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): def test_handles_migration_error_gracefully(self, tmp_path):
"""Should catch exceptions and return False.""" """Should catch exceptions and return False."""