feat: add post-migration cleanup for OpenClaw directories (#4100)

After migrating from OpenClaw, leftover workspace directories contain
state files (todo.json, sessions, logs) that confuse the agent — it
discovers them and reads/writes to stale locations instead of the
Hermes state directory, causing issues like cron jobs reading a
different todo list than interactive sessions.

Changes:
- hermes claw migrate now offers to archive the source directory after
  successful migration (rename to .pre-migration, not delete)
- New `hermes claw cleanup` subcommand for users who already migrated
  and need to archive leftover OpenClaw directories
- Migration notes updated with explicit cleanup guidance
- 42 tests covering all new functionality

Reported by SteveSkedasticity — multiple todo.json files across
~/.hermes/, ~/.openclaw/workspace/, and ~/.openclaw/workspace-assistant/
caused cron jobs to read from wrong locations.

hermes_cli/claw.py

@@ -4,11 +4,15 @@ Usage:
     hermes claw migrate              # Interactive migration from ~/.openclaw
     hermes claw migrate --dry-run    # Preview what would be migrated
     hermes claw migrate --preset full --overwrite  # Full migration, overwrite conflicts
+    hermes claw cleanup              # Archive leftover OpenClaw directories
+    hermes claw cleanup --dry-run    # Preview what would be archived
 """
 
 import importlib.util
 import logging
+import shutil
 import sys
+from datetime import datetime
 from pathlib import Path
 
 from hermes_cli.config import get_hermes_home, get_config_path, load_config, save_config
@@ -20,6 +24,7 @@ from hermes_cli.setup import (
     print_info,
     print_success,
     print_error,
+    print_warning,
     prompt_yes_no,
 )
 
@@ -45,6 +50,18 @@ _OPENCLAW_SCRIPT_INSTALLED = (
     / "openclaw_to_hermes.py"
 )
 
+# Known OpenClaw directory names (current + legacy)
+_OPENCLAW_DIR_NAMES = (".openclaw", ".clawdbot", ".moldbot")
+
+# State files commonly found in OpenClaw workspace directories that cause
+# confusion after migration (the agent discovers them and writes to them)
+_WORKSPACE_STATE_GLOBS = (
+    "*/todo.json",
+    "*/sessions/*",
+    "*/memory/*.json",
+    "*/logs/*",
+)
+
 
 def _find_migration_script() -> Path | None:
     """Find the openclaw_to_hermes.py script in known locations."""
@@ -71,19 +88,88 @@ def _load_migration_module(script_path: Path):
     return mod
 
 
+def _find_openclaw_dirs() -> list[Path]:
+    """Find all OpenClaw directories on disk."""
+    found = []
+    for name in _OPENCLAW_DIR_NAMES:
+        candidate = Path.home() / name
+        if candidate.is_dir():
+            found.append(candidate)
+    return found
+
+
+def _scan_workspace_state(source_dir: Path) -> list[tuple[Path, str]]:
+    """Scan an OpenClaw directory for workspace state files that cause confusion.
+
+    Returns a list of (path, description) tuples.
+    """
+    findings: list[tuple[Path, str]] = []
+
+    # Direct state files in the root
+    for name in ("todo.json", "sessions", "logs"):
+        candidate = source_dir / name
+        if candidate.exists():
+            kind = "directory" if candidate.is_dir() else "file"
+            findings.append((candidate, f"Root {kind}: {name}"))
+
+    # State files inside workspace directories
+    for child in sorted(source_dir.iterdir()):
+        if not child.is_dir() or child.name.startswith("."):
+            continue
+        # Check for workspace-like subdirectories
+        for state_name in ("todo.json", "sessions", "logs", "memory"):
+            state_path = child / state_name
+            if state_path.exists():
+                kind = "directory" if state_path.is_dir() else "file"
+                rel = state_path.relative_to(source_dir)
+                findings.append((state_path, f"Workspace {kind}: {rel}"))
+
+    return findings
+
+
+def _archive_directory(source_dir: Path, dry_run: bool = False) -> Path:
+    """Rename an OpenClaw directory to .pre-migration.
+
+    Returns the archive path.
+    """
+    timestamp = datetime.now().strftime("%Y%m%d")
+    archive_name = f"{source_dir.name}.pre-migration"
+    archive_path = source_dir.parent / archive_name
+
+    # If archive already exists, add timestamp
+    if archive_path.exists():
+        archive_name = f"{source_dir.name}.pre-migration-{timestamp}"
+        archive_path = source_dir.parent / archive_name
+
+    # If still exists (multiple runs same day), add counter
+    counter = 2
+    while archive_path.exists():
+        archive_name = f"{source_dir.name}.pre-migration-{timestamp}-{counter}"
+        archive_path = source_dir.parent / archive_name
+        counter += 1
+
+    if not dry_run:
+        source_dir.rename(archive_path)
+
+    return archive_path
+
+
 def claw_command(args):
     """Route hermes claw subcommands."""
     action = getattr(args, "claw_action", None)
 
     if action == "migrate":
         _cmd_migrate(args)
+    elif action in ("cleanup", "clean"):
+        _cmd_cleanup(args)
     else:
-        print("Usage: hermes claw migrate [options]")
+        print("Usage: hermes claw <command> [options]")
         print()
         print("Commands:")
         print("  migrate          Migrate settings from OpenClaw to Hermes")
+        print("  cleanup          Archive leftover OpenClaw directories after migration")
         print()
-        print("Run 'hermes claw migrate --help' for migration options.")
+        print("Run 'hermes claw <command> --help' for options.")
 
 
 def _cmd_migrate(args):
@@ -210,6 +296,168 @@ def _cmd_migrate(args):
     # Print results
     _print_migration_report(report, dry_run)
 
+    # After successful non-dry-run migration, offer to archive the source directory
+    if not dry_run and report.get("summary", {}).get("migrated", 0) > 0:
+        _offer_source_archival(source_dir, getattr(args, "yes", False))
+
+
+def _offer_source_archival(source_dir: Path, auto_yes: bool = False):
+    """After migration, offer to rename the source directory to prevent state fragmentation.
+
+    OpenClaw workspace directories contain state files (todo.json, sessions, etc.)
+    that the agent may discover and write to, causing confusion. Renaming the
+    directory prevents this.
+    """
+    if not source_dir.is_dir():
+        return
+
+    # Scan for state files that could cause problems
+    state_files = _scan_workspace_state(source_dir)
+
+    print()
+    print_header("Post-Migration Cleanup")
+    print_info("The OpenClaw directory still exists and contains workspace state files")
+    print_info("that can confuse the agent (todo lists, sessions, logs).")
+    if state_files:
+        print()
+        print(color("  Found state files:", Colors.YELLOW))
+        # Show up to 10 most relevant findings
+        for path, desc in state_files[:10]:
+            print(f"      {desc}")
+        if len(state_files) > 10:
+            print(f"      ... and {len(state_files) - 10} more")
+    print()
+    print_info(f"Recommend: rename {source_dir.name}/ to {source_dir.name}.pre-migration/")
+    print_info("This prevents the agent from discovering old workspace directories.")
+    print_info("You can always rename it back if needed.")
+    print()
+
+    if auto_yes or prompt_yes_no(f"Archive {source_dir} now?", default=True):
+        try:
+            archive_path = _archive_directory(source_dir)
+            print_success(f"Archived: {source_dir} → {archive_path}")
+            print_info("The original directory has been renamed, not deleted.")
+            print_info(f"To undo: mv {archive_path} {source_dir}")
+        except OSError as e:
+            print_error(f"Could not archive: {e}")
+            print_info(f"You can do it manually: mv {source_dir} {source_dir}.pre-migration")
+    else:
+        print_info("Skipped. You can archive later with: hermes claw cleanup")
+
+
+def _cmd_cleanup(args):
+    """Archive leftover OpenClaw directories after migration.
+
+    Scans for OpenClaw directories that still exist after migration and offers
+    to rename them to .pre-migration to prevent state fragmentation.
+    """
+    dry_run = getattr(args, "dry_run", False)
+    auto_yes = getattr(args, "yes", False)
+    explicit_source = getattr(args, "source", None)
+
+    print()
+    print(
+        color(
+            "┌─────────────────────────────────────────────────────────┐",
+            Colors.MAGENTA,
+        )
+    )
+    print(
+        color(
+            "│          ⚕ Hermes — OpenClaw Cleanup                   │",
+            Colors.MAGENTA,
+        )
+    )
+    print(
+        color(
+            "└─────────────────────────────────────────────────────────┘",
+            Colors.MAGENTA,
+        )
+    )
+
+    # Find OpenClaw directories
+    if explicit_source:
+        dirs_to_check = [Path(explicit_source)]
+    else:
+        dirs_to_check = _find_openclaw_dirs()
+
+    if not dirs_to_check:
+        print()
+        print_success("No OpenClaw directories found. Nothing to clean up.")
+        return
+
+    total_archived = 0
+
+    for source_dir in dirs_to_check:
+        print()
+        print_header(f"Found: {source_dir}")
+
+        # Scan for state files
+        state_files = _scan_workspace_state(source_dir)
+
+        # Show directory stats
+        try:
+            workspace_dirs = [
+                d for d in source_dir.iterdir()
+                if d.is_dir() and not d.name.startswith(".")
+                and any((d / name).exists() for name in ("todo.json", "SOUL.md", "MEMORY.md", "USER.md"))
+            ]
+        except OSError:
+            workspace_dirs = []
+
+        if workspace_dirs:
+            print_info(f"Workspace directories: {len(workspace_dirs)}")
+            for ws in workspace_dirs[:5]:
+                items = []
+                if (ws / "todo.json").exists():
+                    items.append("todo.json")
+                if (ws / "sessions").is_dir():
+                    items.append("sessions/")
+                if (ws / "SOUL.md").exists():
+                    items.append("SOUL.md")
+                if (ws / "MEMORY.md").exists():
+                    items.append("MEMORY.md")
+                detail = ", ".join(items) if items else "empty"
+                print(f"      {ws.name}/  ({detail})")
+            if len(workspace_dirs) > 5:
+                print(f"      ... and {len(workspace_dirs) - 5} more")
+
+        if state_files:
+            print()
+            print(color(f"  {len(state_files)} state file(s) that could cause confusion:", Colors.YELLOW))
+            for path, desc in state_files[:8]:
+                print(f"      {desc}")
+            if len(state_files) > 8:
+                print(f"      ... and {len(state_files) - 8} more")
+
+        print()
+
+        if dry_run:
+            archive_path = _archive_directory(source_dir, dry_run=True)
+            print_info(f"Would archive: {source_dir} → {archive_path}")
+        else:
+            if auto_yes or prompt_yes_no(f"Archive {source_dir}?", default=True):
+                try:
+                    archive_path = _archive_directory(source_dir)
+                    print_success(f"Archived: {source_dir} → {archive_path}")
+                    total_archived += 1
+                except OSError as e:
+                    print_error(f"Could not archive: {e}")
+                    print_info(f"Try manually: mv {source_dir} {source_dir}.pre-migration")
+            else:
+                print_info("Skipped.")
+
+    # Summary
+    print()
+    if dry_run:
+        print_info(f"Dry run complete. {len(dirs_to_check)} directory(ies) would be archived.")
+        print_info("Run without --dry-run to archive them.")
+    elif total_archived:
+        print_success(f"Cleaned up {total_archived} OpenClaw directory(ies).")
+        print_info("Directories were renamed, not deleted. You can undo by renaming them back.")
+    else:
+        print_info("No directories were archived.")
+
 
 def _print_migration_report(report: dict, dry_run: bool):
     """Print a formatted migration report."""

hermes_cli/main.py

@@ -4712,6 +4712,28 @@ For more help on a command:
         help="Skip confirmation prompts"
     )
 
+    # claw cleanup
+    claw_cleanup = claw_subparsers.add_parser(
+        "cleanup",
+        aliases=["clean"],
+        help="Archive leftover OpenClaw directories after migration",
+        description="Scan for and archive leftover OpenClaw directories to prevent state fragmentation"
+    )
+    claw_cleanup.add_argument(
+        "--source",
+        help="Path to a specific OpenClaw directory to clean up"
+    )
+    claw_cleanup.add_argument(
+        "--dry-run",
+        action="store_true",
+        help="Preview what would be archived without making changes"
+    )
+    claw_cleanup.add_argument(
+        "--yes", "-y",
+        action="store_true",
+        help="Skip confirmation prompts"
+    )
+
     def cmd_claw(args):
         from hermes_cli.claw import claw_command
         claw_command(args)

optional-skills/migration/openclaw-migration/scripts/openclaw_to_hermes.py

@@ -2455,9 +2455,24 @@ class Migrator:
             notes.append("")
 
         notes.extend([
+            "## IMPORTANT: Archive the OpenClaw Directory",
+            "",
+            "After migration, your OpenClaw directory still exists on disk with workspace",
+            "state files (todo.json, sessions, logs). If the Hermes agent discovers these",
+            "directories, it may read/write to them instead of the Hermes state, causing",
+            "confusion (e.g., cron jobs reading a different todo list than interactive sessions).",
+            "",
+            "**Strongly recommended:** Run `hermes claw cleanup` to rename the OpenClaw",
+            "directory to `.openclaw.pre-migration`. This prevents the agent from finding it.",
+            "The directory is renamed, not deleted — you can undo this at any time.",
+            "",
+            "If you skip this step and notice the agent getting confused about workspaces",
+            "or todo lists, run `hermes claw cleanup` to fix it.",
+            "",
             "## Hermes-Specific Setup",
             "",
             "After migration, you may want to:",
+            "- Run `hermes claw cleanup` to archive the OpenClaw directory (prevents state confusion)",
             "- Run `hermes setup` to configure any remaining settings",
             "- Run `hermes mcp list` to verify MCP servers were imported correctly",
             "- Run `hermes cron` to recreate scheduled tasks (see archive/cron-config.json)",

tests/hermes_cli/test_claw.py

@@ -40,6 +40,119 @@ class TestFindMigrationScript:
             assert claw_mod._find_migration_script() is None
 
 
+# ---------------------------------------------------------------------------
+# _find_openclaw_dirs
+# ---------------------------------------------------------------------------
+
+
+class TestFindOpenclawDirs:
+    """Test discovery of OpenClaw directories."""
+
+    def test_finds_openclaw_dir(self, tmp_path):
+        openclaw = tmp_path / ".openclaw"
+        openclaw.mkdir()
+        with patch("pathlib.Path.home", return_value=tmp_path):
+            found = claw_mod._find_openclaw_dirs()
+        assert openclaw in found
+
+    def test_finds_legacy_dirs(self, tmp_path):
+        clawdbot = tmp_path / ".clawdbot"
+        clawdbot.mkdir()
+        moldbot = tmp_path / ".moldbot"
+        moldbot.mkdir()
+        with patch("pathlib.Path.home", return_value=tmp_path):
+            found = claw_mod._find_openclaw_dirs()
+        assert len(found) == 2
+        assert clawdbot in found
+        assert moldbot in found
+
+    def test_returns_empty_when_none_exist(self, tmp_path):
+        with patch("pathlib.Path.home", return_value=tmp_path):
+            found = claw_mod._find_openclaw_dirs()
+        assert found == []
+
+
+# ---------------------------------------------------------------------------
+# _scan_workspace_state
+# ---------------------------------------------------------------------------
+
+
+class TestScanWorkspaceState:
+    """Test scanning for workspace state files."""
+
+    def test_finds_root_state_files(self, tmp_path):
+        (tmp_path / "todo.json").write_text("{}")
+        (tmp_path / "sessions").mkdir()
+        findings = claw_mod._scan_workspace_state(tmp_path)
+        descs = [desc for _, desc in findings]
+        assert any("todo.json" in d for d in descs)
+        assert any("sessions" in d for d in descs)
+
+    def test_finds_workspace_state_files(self, tmp_path):
+        ws = tmp_path / "workspace"
+        ws.mkdir()
+        (ws / "todo.json").write_text("{}")
+        (ws / "sessions").mkdir()
+        findings = claw_mod._scan_workspace_state(tmp_path)
+        descs = [desc for _, desc in findings]
+        assert any("workspace/todo.json" in d for d in descs)
+        assert any("workspace/sessions" in d for d in descs)
+
+    def test_ignores_hidden_dirs(self, tmp_path):
+        scan_dir = tmp_path / "scan_target"
+        scan_dir.mkdir()
+        hidden = scan_dir / ".git"
+        hidden.mkdir()
+        (hidden / "todo.json").write_text("{}")
+        findings = claw_mod._scan_workspace_state(scan_dir)
+        assert len(findings) == 0
+
+    def test_empty_dir_returns_empty(self, tmp_path):
+        scan_dir = tmp_path / "scan_target"
+        scan_dir.mkdir()
+        findings = claw_mod._scan_workspace_state(scan_dir)
+        assert findings == []
+
+
+# ---------------------------------------------------------------------------
+# _archive_directory
+# ---------------------------------------------------------------------------
+
+
+class TestArchiveDirectory:
+    """Test directory archival (rename)."""
+
+    def test_renames_to_pre_migration(self, tmp_path):
+        source = tmp_path / ".openclaw"
+        source.mkdir()
+        (source / "test.txt").write_text("data")
+
+        archive_path = claw_mod._archive_directory(source)
+        assert archive_path == tmp_path / ".openclaw.pre-migration"
+        assert archive_path.is_dir()
+        assert not source.exists()
+        assert (archive_path / "test.txt").read_text() == "data"
+
+    def test_adds_timestamp_when_archive_exists(self, tmp_path):
+        source = tmp_path / ".openclaw"
+        source.mkdir()
+        # Pre-existing archive
+        (tmp_path / ".openclaw.pre-migration").mkdir()
+
+        archive_path = claw_mod._archive_directory(source)
+        assert ".pre-migration-" in archive_path.name
+        assert archive_path.is_dir()
+        assert not source.exists()
+
+    def test_dry_run_does_not_rename(self, tmp_path):
+        source = tmp_path / ".openclaw"
+        source.mkdir()
+
+        archive_path = claw_mod._archive_directory(source, dry_run=True)
+        assert archive_path == tmp_path / ".openclaw.pre-migration"
+        assert source.is_dir()  # Still exists
+
+
 # ---------------------------------------------------------------------------
 # claw_command routing
 # ---------------------------------------------------------------------------
@@ -56,11 +169,24 @@ class TestClawCommand:
             claw_mod.claw_command(args)
         mock.assert_called_once_with(args)
 
+    def test_routes_to_cleanup(self):
+        args = Namespace(claw_action="cleanup", source=None, dry_run=False, yes=False)
+        with patch.object(claw_mod, "_cmd_cleanup") as mock:
+            claw_mod.claw_command(args)
+        mock.assert_called_once_with(args)
+
+    def test_routes_clean_alias(self):
+        args = Namespace(claw_action="clean", source=None, dry_run=False, yes=False)
+        with patch.object(claw_mod, "_cmd_cleanup") as mock:
+            claw_mod.claw_command(args)
+        mock.assert_called_once_with(args)
+
     def test_shows_help_for_no_action(self, capsys):
         args = Namespace(claw_action=None)
         claw_mod.claw_command(args)
         captured = capsys.readouterr()
         assert "migrate" in captured.out
+        assert "cleanup" in captured.out
 
 
 # ---------------------------------------------------------------------------
@@ -168,6 +294,7 @@ class TestCmdMigrate:
             patch.object(claw_mod, "_load_migration_module", return_value=fake_mod),
             patch.object(claw_mod, "get_config_path", return_value=config_path),
             patch.object(claw_mod, "prompt_yes_no", return_value=True),
+            patch.object(claw_mod, "_offer_source_archival"),
         ):
             claw_mod._cmd_migrate(args)
 
@@ -175,6 +302,75 @@ class TestCmdMigrate:
         assert "Migration Results" in captured.out
         assert "Migration complete!" in captured.out
 
+    def test_execute_offers_archival_on_success(self, tmp_path, capsys):
+        """After successful migration, _offer_source_archival should be called."""
+        openclaw_dir = tmp_path / ".openclaw"
+        openclaw_dir.mkdir()
+
+        fake_mod = ModuleType("openclaw_to_hermes")
+        fake_mod.resolve_selected_options = MagicMock(return_value={"soul"})
+        fake_migrator = MagicMock()
+        fake_migrator.migrate.return_value = {
+            "summary": {"migrated": 3, "skipped": 0, "conflict": 0, "error": 0},
+            "items": [
+                {"kind": "soul", "status": "migrated", "destination": str(tmp_path / "SOUL.md")},
+            ],
+        }
+        fake_mod.Migrator = MagicMock(return_value=fake_migrator)
+
+        args = Namespace(
+            source=str(openclaw_dir),
+            dry_run=False, preset="full", overwrite=False,
+            migrate_secrets=False, workspace_target=None,
+            skill_conflict="skip", yes=True,
+        )
+
+        with (
+            patch.object(claw_mod, "_find_migration_script", return_value=tmp_path / "s.py"),
+            patch.object(claw_mod, "_load_migration_module", return_value=fake_mod),
+            patch.object(claw_mod, "get_config_path", return_value=tmp_path / "config.yaml"),
+            patch.object(claw_mod, "save_config"),
+            patch.object(claw_mod, "load_config", return_value={}),
+            patch.object(claw_mod, "_offer_source_archival") as mock_archival,
+        ):
+            claw_mod._cmd_migrate(args)
+
+        mock_archival.assert_called_once_with(openclaw_dir, True)
+
+    def test_dry_run_skips_archival(self, tmp_path, capsys):
+        """Dry run should not offer archival."""
+        openclaw_dir = tmp_path / ".openclaw"
+        openclaw_dir.mkdir()
+
+        fake_mod = ModuleType("openclaw_to_hermes")
+        fake_mod.resolve_selected_options = MagicMock(return_value=set())
+        fake_migrator = MagicMock()
+        fake_migrator.migrate.return_value = {
+            "summary": {"migrated": 2, "skipped": 0, "conflict": 0, "error": 0},
+            "items": [],
+            "preset": "full",
+        }
+        fake_mod.Migrator = MagicMock(return_value=fake_migrator)
+
+        args = Namespace(
+            source=str(openclaw_dir),
+            dry_run=True, preset="full", overwrite=False,
+            migrate_secrets=False, workspace_target=None,
+            skill_conflict="skip", yes=False,
+        )
+
+        with (
+            patch.object(claw_mod, "_find_migration_script", return_value=tmp_path / "s.py"),
+            patch.object(claw_mod, "_load_migration_module", return_value=fake_mod),
+            patch.object(claw_mod, "get_config_path", return_value=tmp_path / "config.yaml"),
+            patch.object(claw_mod, "save_config"),
+            patch.object(claw_mod, "load_config", return_value={}),
+            patch.object(claw_mod, "_offer_source_archival") as mock_archival,
+        ):
+            claw_mod._cmd_migrate(args)
+
+        mock_archival.assert_not_called()
+
     def test_execute_cancelled_by_user(self, tmp_path, capsys):
         openclaw_dir = tmp_path / ".openclaw"
         openclaw_dir.mkdir()
@@ -290,6 +486,172 @@ class TestCmdMigrate:
         assert call_kwargs["migrate_secrets"] is True
 
 
+# ---------------------------------------------------------------------------
+# _offer_source_archival
+# ---------------------------------------------------------------------------
+
+
+class TestOfferSourceArchival:
+    """Test the post-migration archival offer."""
+
+    def test_archives_with_auto_yes(self, tmp_path, capsys):
+        source = tmp_path / ".openclaw"
+        source.mkdir()
+        (source / "workspace").mkdir()
+        (source / "workspace" / "todo.json").write_text("{}")
+
+        claw_mod._offer_source_archival(source, auto_yes=True)
+
+        captured = capsys.readouterr()
+        assert "Archived" in captured.out
+        assert not source.exists()
+        assert (tmp_path / ".openclaw.pre-migration").is_dir()
+
+    def test_skips_when_user_declines(self, tmp_path, capsys):
+        source = tmp_path / ".openclaw"
+        source.mkdir()
+
+        with patch.object(claw_mod, "prompt_yes_no", return_value=False):
+            claw_mod._offer_source_archival(source, auto_yes=False)
+
+        captured = capsys.readouterr()
+        assert "Skipped" in captured.out
+        assert source.is_dir()  # Still exists
+
+    def test_noop_when_source_missing(self, tmp_path, capsys):
+        claw_mod._offer_source_archival(tmp_path / "nonexistent", auto_yes=True)
+        captured = capsys.readouterr()
+        assert captured.out == ""  # No output
+
+    def test_shows_state_files(self, tmp_path, capsys):
+        source = tmp_path / ".openclaw"
+        source.mkdir()
+        ws = source / "workspace"
+        ws.mkdir()
+        (ws / "todo.json").write_text("{}")
+
+        with patch.object(claw_mod, "prompt_yes_no", return_value=False):
+            claw_mod._offer_source_archival(source, auto_yes=False)
+
+        captured = capsys.readouterr()
+        assert "todo.json" in captured.out
+
+    def test_handles_archive_error(self, tmp_path, capsys):
+        source = tmp_path / ".openclaw"
+        source.mkdir()
+
+        with patch.object(claw_mod, "_archive_directory", side_effect=OSError("permission denied")):
+            claw_mod._offer_source_archival(source, auto_yes=True)
+
+        captured = capsys.readouterr()
+        assert "Could not archive" in captured.out
+
+
+# ---------------------------------------------------------------------------
+# _cmd_cleanup
+# ---------------------------------------------------------------------------
+
+
+class TestCmdCleanup:
+    """Test the cleanup command handler."""
+
+    def test_no_dirs_found(self, tmp_path, capsys):
+        args = Namespace(source=None, dry_run=False, yes=False)
+        with patch.object(claw_mod, "_find_openclaw_dirs", return_value=[]):
+            claw_mod._cmd_cleanup(args)
+        captured = capsys.readouterr()
+        assert "No OpenClaw directories found" in captured.out
+
+    def test_dry_run_lists_dirs(self, tmp_path, capsys):
+        openclaw = tmp_path / ".openclaw"
+        openclaw.mkdir()
+        ws = openclaw / "workspace"
+        ws.mkdir()
+        (ws / "todo.json").write_text("{}")
+
+        args = Namespace(source=None, dry_run=True, yes=False)
+        with patch.object(claw_mod, "_find_openclaw_dirs", return_value=[openclaw]):
+            claw_mod._cmd_cleanup(args)
+
+        captured = capsys.readouterr()
+        assert "Would archive" in captured.out
+        assert openclaw.is_dir()  # Not actually archived
+
+    def test_archives_with_yes(self, tmp_path, capsys):
+        openclaw = tmp_path / ".openclaw"
+        openclaw.mkdir()
+        (openclaw / "workspace").mkdir()
+        (openclaw / "workspace" / "todo.json").write_text("{}")
+
+        args = Namespace(source=None, dry_run=False, yes=True)
+        with patch.object(claw_mod, "_find_openclaw_dirs", return_value=[openclaw]):
+            claw_mod._cmd_cleanup(args)
+
+        captured = capsys.readouterr()
+        assert "Archived" in captured.out
+        assert "Cleaned up 1" in captured.out
+        assert not openclaw.exists()
+        assert (tmp_path / ".openclaw.pre-migration").is_dir()
+
+    def test_skips_when_user_declines(self, tmp_path, capsys):
+        openclaw = tmp_path / ".openclaw"
+        openclaw.mkdir()
+
+        args = Namespace(source=None, dry_run=False, yes=False)
+        with (
+            patch.object(claw_mod, "_find_openclaw_dirs", return_value=[openclaw]),
+            patch.object(claw_mod, "prompt_yes_no", return_value=False),
+        ):
+            claw_mod._cmd_cleanup(args)
+
+        captured = capsys.readouterr()
+        assert "Skipped" in captured.out
+        assert openclaw.is_dir()
+
+    def test_explicit_source(self, tmp_path, capsys):
+        custom_dir = tmp_path / "my-openclaw"
+        custom_dir.mkdir()
+        (custom_dir / "todo.json").write_text("{}")
+
+        args = Namespace(source=str(custom_dir), dry_run=False, yes=True)
+        claw_mod._cmd_cleanup(args)
+
+        captured = capsys.readouterr()
+        assert "Archived" in captured.out
+        assert not custom_dir.exists()
+
+    def test_shows_workspace_details(self, tmp_path, capsys):
+        openclaw = tmp_path / ".openclaw"
+        openclaw.mkdir()
+        ws = openclaw / "workspace"
+        ws.mkdir()
+        (ws / "todo.json").write_text("{}")
+        (ws / "SOUL.md").write_text("# Soul")
+
+        args = Namespace(source=None, dry_run=True, yes=False)
+        with patch.object(claw_mod, "_find_openclaw_dirs", return_value=[openclaw]):
+            claw_mod._cmd_cleanup(args)
+
+        captured = capsys.readouterr()
+        assert "workspace/" in captured.out
+        assert "todo.json" in captured.out
+
+    def test_handles_multiple_dirs(self, tmp_path, capsys):
+        openclaw = tmp_path / ".openclaw"
+        openclaw.mkdir()
+        clawdbot = tmp_path / ".clawdbot"
+        clawdbot.mkdir()
+
+        args = Namespace(source=None, dry_run=False, yes=True)
+        with patch.object(claw_mod, "_find_openclaw_dirs", return_value=[openclaw, clawdbot]):
+            claw_mod._cmd_cleanup(args)
+
+        captured = capsys.readouterr()
+        assert "Cleaned up 2" in captured.out
+        assert not openclaw.exists()
+        assert not clawdbot.exists()
+
+
 # ---------------------------------------------------------------------------
 # _print_migration_report
 # ---------------------------------------------------------------------------