refactor(hermes_cli): collapse claw confirm/report printing into shared helpers
This commit is contained in:
Teknium
+151
-212
@@ -12,9 +12,8 @@ from typing import Callable, Optional
|
||||
|
||||
from hermes_cli.config import get_hermes_home, get_config_path, load_config, save_config
|
||||
from hermes_constants import get_optional_skills_dir
|
||||
from hermes_cli.setup import (
|
||||
Colors, color, print_header, print_info, print_success, print_error, prompt_yes_no,
|
||||
)
|
||||
from hermes_cli.setup import (Colors, color, print_header, print_info, print_success, print_error,
|
||||
prompt_yes_no)
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
@@ -28,9 +27,8 @@ _OPENCLAW_SCRIPT_INSTALLED = get_hermes_home() / "skills" / _SCRIPT_REL
|
||||
# Known OpenClaw directory names (current + legacy)
|
||||
_OPENCLAW_DIR_NAMES = (".openclaw", ".clawdbot", ".moltbot")
|
||||
|
||||
# `hermes claw migrate` flags and their defaults. Secrets are never included implicitly — they
|
||||
# must be requested via --migrate-secrets even under --preset full (mirrors OpenClaw's two-phase
|
||||
# migrate-hermes posture), so a full run cannot silently import API keys.
|
||||
# `hermes claw migrate` flags/defaults. Secrets are never included implicitly: --migrate-secrets
|
||||
# is required even under --preset full (OpenClaw's two-phase posture); no silent API-key import.
|
||||
_MIGRATE_ARG_DEFAULTS = (
|
||||
("source", None), ("dry_run", False), ("preset", "full"), ("overwrite", False),
|
||||
("migrate_secrets", False), ("workspace_target", None), ("skill_conflict", "skip"),
|
||||
@@ -44,8 +42,11 @@ _REPORT_REASON_GROUPS = (
|
||||
("skipped", " ─ Skipped:", Colors.DIM, ""),
|
||||
("error", " ✗ Errors:", Colors.RED, "unknown error"),
|
||||
)
|
||||
# Summary-line counters after the migrated count: (summary key, label).
|
||||
_SUMMARY_COUNT_LABELS = (("conflict", "conflict(s)"), ("skipped", "skipped"), ("error", "error(s)"))
|
||||
|
||||
# Workspace marker files listed by `hermes claw cleanup` (name, display label, presence check).
|
||||
# A subdir with any of these files is a workspace; the labelled subset is listed by cleanup.
|
||||
_WORKSPACE_MARKERS = ("todo.json", "SOUL.md", "MEMORY.md", "USER.md")
|
||||
_WORKSPACE_ITEM_LABELS = (
|
||||
("todo.json", "todo.json", Path.exists), ("sessions", "sessions/", Path.is_dir),
|
||||
("SOUL.md", "SOUL.md", Path.exists), ("MEMORY.md", "MEMORY.md", Path.exists),
|
||||
@@ -60,52 +61,51 @@ def _print_banner(title: str) -> None:
|
||||
print(color(line, Colors.MAGENTA))
|
||||
|
||||
|
||||
def _error_block(headline: str, *lines: str) -> None:
|
||||
"""Print a blank line, an error headline, then each line as info."""
|
||||
def _info(*lines: str) -> None:
|
||||
"""print_info each line; an empty string prints a bare blank line instead."""
|
||||
for line in lines:
|
||||
print_info(line) if line else print()
|
||||
|
||||
|
||||
def _error_block(headline: str, *lines: str, debug: Optional[str] = None) -> None:
|
||||
"""Print a blank line, an error headline, then info lines; ``debug`` logs the traceback."""
|
||||
print()
|
||||
print_error(headline)
|
||||
for line in lines:
|
||||
print_info(line)
|
||||
_info(*lines)
|
||||
if debug:
|
||||
logger.debug(debug, exc_info=True)
|
||||
|
||||
|
||||
def _confirm(auto_yes: bool, question: str, *, default: bool, declined: str,
|
||||
non_tty: Optional[tuple[str, ...]] = None) -> Optional[bool]:
|
||||
"""Ask to proceed unless --yes; print ``declined`` and return False when the user says no.
|
||||
|
||||
With ``non_tty`` set, a non-interactive stdin prints those lines and returns None instead of
|
||||
prompting (callers treat None as "don't apply" but never as an explicit refusal).
|
||||
"""
|
||||
"""Ask to proceed unless --yes; print ``declined`` and return False on an explicit no. With
|
||||
``non_tty`` set, a non-interactive stdin prints those lines and returns None (not a refusal)."""
|
||||
if auto_yes:
|
||||
return True
|
||||
if non_tty is not None and not sys.stdin.isatty():
|
||||
for line in non_tty:
|
||||
print_info(line)
|
||||
_info(*non_tty)
|
||||
return None
|
||||
if prompt_yes_no(question, default=default):
|
||||
return True
|
||||
print_info(declined)
|
||||
return False
|
||||
answer = prompt_yes_no(question, default=default)
|
||||
if not answer:
|
||||
print_info(declined)
|
||||
return answer
|
||||
|
||||
|
||||
def _warn_token_conflict(auto_yes: bool, headline: str, lines: list[str], declined: str,
|
||||
non_tty: Optional[tuple[str, ...]] = None) -> None:
|
||||
"""Print a bot-token conflict warning and exit 0 if the user explicitly declines to continue."""
|
||||
_error_block(headline, *lines)
|
||||
def _warn_running(auto_yes: bool, headline: str, running: list[str], lines: tuple[str, ...],
|
||||
question: str, declined: str, non_tty: Optional[tuple[str, ...]] = None):
|
||||
"""Print the 'still running' block (headline, `* detail` per process, advice) then _confirm."""
|
||||
_error_block(headline, *(f" * {detail}" for detail in running), *lines)
|
||||
print()
|
||||
if _confirm(auto_yes, "Continue anyway?", default=False, declined=declined,
|
||||
non_tty=non_tty) is False:
|
||||
sys.exit(0)
|
||||
return _confirm(auto_yes, question, default=False, declined=declined, non_tty=non_tty)
|
||||
|
||||
|
||||
def _detect_openclaw_processes() -> list[str]:
|
||||
"""Detect running OpenClaw processes and services."""
|
||||
found: list[str] = []
|
||||
if sys.platform == "win32":
|
||||
# bounded_probe_run: a plain subprocess.run(timeout=...) can hang forever on Windows in
|
||||
# post-timeout cleanup when a conhost.exe descendant holds duplicated pipe handles — and
|
||||
# a hang is not an exception, so the try/except here can't save the caller.
|
||||
# bounded_probe_run: plain subprocess.run(timeout=...) can hang forever on Windows when a
|
||||
# conhost.exe descendant holds duplicated pipe handles — a hang is not an exception.
|
||||
from hermes_cli._subprocess_compat import bounded_probe_run
|
||||
|
||||
try:
|
||||
for exe in ("openclaw.exe", "clawd.exe"):
|
||||
result = bounded_probe_run(["tasklist", "/FI", f"IMAGENAME eq {exe}"], timeout=5)
|
||||
@@ -118,8 +118,8 @@ def _detect_openclaw_processes() -> list[str]:
|
||||
'Select-Object -First 1 ProcessId'
|
||||
)
|
||||
result = bounded_probe_run(["powershell", "-NoProfile", "-Command", ps_cmd], timeout=5)
|
||||
if result is not None and (result.stdout or "").strip():
|
||||
pid = result.stdout.strip()
|
||||
pid = (result.stdout or "").strip() if result is not None else ""
|
||||
if pid:
|
||||
found.append(f"node.exe process with openclaw in command line (PID {pid})")
|
||||
except Exception:
|
||||
pass
|
||||
@@ -145,37 +145,34 @@ def _posix_probe(cmd: list[str], timeout: int):
|
||||
def _warn_if_openclaw_running(auto_yes: bool) -> None:
|
||||
"""Warn if OpenClaw is still running: Telegram/Discord/Slack allow one session per bot token."""
|
||||
running = _detect_openclaw_processes()
|
||||
if running:
|
||||
_warn_token_conflict(
|
||||
auto_yes, "OpenClaw appears to be running:",
|
||||
[f" * {detail}" for detail in running] + [
|
||||
"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.",
|
||||
"Recommendation: stop OpenClaw before migrating."],
|
||||
declined="Migration cancelled. Stop OpenClaw and try again.",
|
||||
non_tty=("Non-interactive session — continuing to preview only.",),
|
||||
)
|
||||
if running and _warn_running(
|
||||
auto_yes, "OpenClaw appears to be running:", running,
|
||||
("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.",
|
||||
"Recommendation: stop OpenClaw before migrating."),
|
||||
"Continue anyway?", declined="Migration cancelled. Stop OpenClaw and try again.",
|
||||
non_tty=("Non-interactive session — continuing to preview only.",),
|
||||
) is False:
|
||||
sys.exit(0)
|
||||
|
||||
|
||||
def _warn_if_gateway_running(auto_yes: bool) -> None:
|
||||
"""Warn if a Hermes gateway has connected platforms (token conflicts, e.g. Telegram 409)."""
|
||||
from gateway.status import get_running_pid, read_runtime_status
|
||||
|
||||
if not get_running_pid():
|
||||
return
|
||||
platforms = (read_runtime_status() or {}).get("platforms") or {}
|
||||
platforms = ((read_runtime_status() or {}).get("platforms") or {}) if get_running_pid() else {}
|
||||
connected = [name for name, info in platforms.items()
|
||||
if isinstance(info, dict) and info.get("state") == "connected"]
|
||||
if connected:
|
||||
_warn_token_conflict(
|
||||
auto_yes, "Hermes gateway is running with active connections: " + ", ".join(connected),
|
||||
["Migrating bot tokens while the gateway is active will cause "
|
||||
"conflicts (Telegram, Discord, and Slack only allow one active "
|
||||
"session per token).",
|
||||
"Recommendation: stop the gateway first with 'hermes gateway stop'."],
|
||||
declined="Migration cancelled. Stop the gateway and try again.",
|
||||
)
|
||||
if connected and _warn_running(
|
||||
auto_yes, "Hermes gateway is running with active connections: " + ", ".join(connected), [],
|
||||
("Migrating bot tokens while the gateway is active will cause "
|
||||
"conflicts (Telegram, Discord, and Slack only allow one active "
|
||||
"session per token).",
|
||||
"Recommendation: stop the gateway first with 'hermes gateway stop'."),
|
||||
"Continue anyway?", declined="Migration cancelled. Stop the gateway and try again.",
|
||||
) is False:
|
||||
sys.exit(0)
|
||||
|
||||
|
||||
def _find_migration_script() -> Path | None:
|
||||
@@ -188,9 +185,8 @@ def _load_migration_module(script_path: Path):
|
||||
spec = importlib.util.spec_from_file_location("openclaw_to_hermes", script_path)
|
||||
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).
|
||||
sys.modules[spec.name] = mod
|
||||
mod = sys.modules[spec.name] = importlib.util.module_from_spec(spec)
|
||||
try:
|
||||
spec.loader.exec_module(mod)
|
||||
except Exception:
|
||||
@@ -211,16 +207,11 @@ def _scan_workspace_state(source_dir: Path) -> list[tuple[Path, str]]:
|
||||
children = sorted(source_dir.iterdir())
|
||||
except OSError:
|
||||
children = []
|
||||
candidates += [
|
||||
(child / name, "Workspace")
|
||||
for child in children if child.is_dir() and not child.name.startswith(".")
|
||||
for name in ("todo.json", "sessions", "logs", "memory")
|
||||
]
|
||||
return [
|
||||
(p, f"{scope} {'directory' if p.is_dir() else 'file'}: "
|
||||
f"{p.relative_to(source_dir).as_posix()}")
|
||||
for p, scope in candidates if p.exists()
|
||||
]
|
||||
candidates += [(child / name, "Workspace") for child in children
|
||||
if child.is_dir() and not child.name.startswith(".")
|
||||
for name in ("todo.json", "sessions", "logs", "memory")]
|
||||
return [(p, f"{scope} {'directory' if p.is_dir() else 'file'}: "
|
||||
f"{p.relative_to(source_dir).as_posix()}") for p, scope in candidates if p.exists()]
|
||||
|
||||
|
||||
def _archive_directory(source_dir: Path, dry_run: bool = False) -> Path:
|
||||
@@ -255,33 +246,26 @@ def _cmd_migrate(args):
|
||||
opts.source_dir = (Path(opts.source) if opts.source
|
||||
else next(iter(_find_openclaw_dirs()), Path.home() / ".openclaw"))
|
||||
_print_banner("OpenClaw Migration")
|
||||
|
||||
if not opts.source_dir.is_dir():
|
||||
_error_block(
|
||||
return _error_block(
|
||||
f"OpenClaw directory not found: {opts.source_dir}",
|
||||
"Make sure your OpenClaw installation is at the expected path.",
|
||||
"You can specify a custom path: hermes claw migrate --source /path/to/.openclaw",
|
||||
)
|
||||
return
|
||||
"You can specify a custom path: hermes claw migrate --source /path/to/.openclaw")
|
||||
script_path = _find_migration_script()
|
||||
if not script_path:
|
||||
_error_block("Migration script not found.", "Expected at one of:", f" {_OPENCLAW_SCRIPT}",
|
||||
f" {_OPENCLAW_SCRIPT_INSTALLED}",
|
||||
"Make sure the openclaw-migration skill is installed.")
|
||||
return
|
||||
|
||||
return _error_block(
|
||||
"Migration script not found.", "Expected at one of:", f" {_OPENCLAW_SCRIPT}",
|
||||
f" {_OPENCLAW_SCRIPT_INSTALLED}",
|
||||
"Make sure the openclaw-migration skill is installed.")
|
||||
opts.hermes_home = get_hermes_home()
|
||||
print()
|
||||
print_header("Migration Settings")
|
||||
print_info(f"Source: {opts.source_dir}")
|
||||
print_info(f"Target: {opts.hermes_home}")
|
||||
print_info(f"Preset: {opts.preset}")
|
||||
print_info(f"Overwrite: {'yes' if opts.overwrite else 'no (skip conflicts)'}")
|
||||
print_info(f"Secrets: {'yes (allowlisted only)' if opts.migrate_secrets else 'no'}")
|
||||
if opts.skill_conflict != "skip":
|
||||
print_info(f"Skill conflicts: {opts.skill_conflict}")
|
||||
if opts.workspace_target:
|
||||
print_info(f"Workspace: {opts.workspace_target}")
|
||||
_info(f"Source: {opts.source_dir}", f"Target: {opts.hermes_home}",
|
||||
f"Preset: {opts.preset}",
|
||||
f"Overwrite: {'yes' if opts.overwrite else 'no (skip conflicts)'}",
|
||||
f"Secrets: {'yes (allowlisted only)' if opts.migrate_secrets else 'no'}",
|
||||
*([f"Skill conflicts: {opts.skill_conflict}"] if opts.skill_conflict != "skip" else []),
|
||||
*([f"Workspace: {opts.workspace_target}"] if opts.workspace_target else []))
|
||||
print()
|
||||
# Migrating tokens while OpenClaw or the gateway is active causes conflicts (e.g. Telegram 409).
|
||||
_warn_if_openclaw_running(opts.yes)
|
||||
@@ -289,7 +273,6 @@ def _cmd_migrate(args):
|
||||
# Ensure config.yaml exists before migration tries to read it
|
||||
if not get_config_path().exists():
|
||||
save_config(load_config())
|
||||
|
||||
run_migrator = _load_migrator(script_path, opts)
|
||||
if run_migrator is None or not _preview_migration(run_migrator, opts):
|
||||
return
|
||||
@@ -306,24 +289,18 @@ def _load_migrator(script_path: Path, opts: SimpleNamespace) -> Optional[Callabl
|
||||
try:
|
||||
mod = _load_migration_module(script_path)
|
||||
except Exception as e:
|
||||
_error_block(f"Could not load migration script: {e}")
|
||||
logger.debug("OpenClaw migration error", exc_info=True)
|
||||
return None
|
||||
return _error_block(f"Could not load migration script: {e}",
|
||||
debug="OpenClaw migration error")
|
||||
if mod is None:
|
||||
print_error("Could not load migration script.")
|
||||
return None
|
||||
return print_error("Could not load migration script.")
|
||||
selected = mod.resolve_selected_options(None, None, preset=opts.preset)
|
||||
ws_target = Path(opts.workspace_target).resolve() if opts.workspace_target else None
|
||||
|
||||
def _run_migrator(execute: bool) -> dict:
|
||||
return mod.Migrator(
|
||||
source_root=opts.source_dir.resolve(), target_root=opts.hermes_home.resolve(),
|
||||
execute=execute, workspace_target=ws_target, overwrite=opts.overwrite,
|
||||
migrate_secrets=opts.migrate_secrets, output_dir=None, selected_options=selected,
|
||||
preset_name=opts.preset, skill_conflict_mode=opts.skill_conflict,
|
||||
).migrate()
|
||||
|
||||
return _run_migrator
|
||||
return lambda execute: mod.Migrator(
|
||||
source_root=opts.source_dir.resolve(), target_root=opts.hermes_home.resolve(),
|
||||
execute=execute, workspace_target=ws_target, overwrite=opts.overwrite,
|
||||
migrate_secrets=opts.migrate_secrets, output_dir=None, selected_options=selected,
|
||||
preset_name=opts.preset, skill_conflict_mode=opts.skill_conflict,
|
||||
).migrate()
|
||||
|
||||
|
||||
def _preview_migration(run_migrator: Callable[[bool], dict], opts: SimpleNamespace) -> bool:
|
||||
@@ -331,113 +308,88 @@ def _preview_migration(run_migrator: Callable[[bool], dict], opts: SimpleNamespa
|
||||
try:
|
||||
preview_report = run_migrator(False)
|
||||
except Exception as e:
|
||||
_error_block(f"Migration preview failed: {e}")
|
||||
logger.debug("OpenClaw migration preview error", exc_info=True)
|
||||
_error_block(f"Migration preview failed: {e}", debug="OpenClaw migration preview error")
|
||||
return False
|
||||
preview_summary = preview_report.get("summary", {})
|
||||
preview_count = preview_summary.get("migrated", 0)
|
||||
preview_conflicts = preview_summary.get("conflict", 0)
|
||||
|
||||
summary = preview_report.get("summary", {})
|
||||
count, conflicts = summary.get("migrated", 0), summary.get("conflict", 0)
|
||||
# "Nothing to migrate" means nothing migrated AND nothing blocked by conflicts. With
|
||||
# conflicts, still show the plan and surface the --overwrite guidance instead of bailing.
|
||||
if preview_count == 0 and preview_conflicts == 0:
|
||||
print()
|
||||
print_info("Nothing to migrate from OpenClaw.")
|
||||
_print_migration_report(preview_report, dry_run=True)
|
||||
return False
|
||||
print()
|
||||
print_header(
|
||||
f"Migration Preview — {preview_count} item(s) would be imported"
|
||||
if preview_count > 0
|
||||
else f"Migration Preview — {preview_conflicts} conflict(s), nothing would be imported"
|
||||
)
|
||||
print_info("No changes have been made yet. Review the list below:")
|
||||
if count == 0 and conflicts == 0:
|
||||
print_info("Nothing to migrate from OpenClaw.")
|
||||
else:
|
||||
what = (f"{count} item(s) would be imported" if count > 0
|
||||
else f"{conflicts} conflict(s), nothing would be imported")
|
||||
print_header(f"Migration Preview — {what}")
|
||||
print_info("No changes have been made yet. Review the list below:")
|
||||
_print_migration_report(preview_report, dry_run=True)
|
||||
if opts.dry_run:
|
||||
if opts.dry_run or (count == 0 and conflicts == 0):
|
||||
return False
|
||||
|
||||
# Modelled on OpenClaw's assertConflictFreePlan(): apply is a safe no-op on conflicts unless
|
||||
# the user opts in to overwriting — otherwise "yes, proceed" would silently skip every
|
||||
# conflicting item.
|
||||
if preview_conflicts > 0 and not opts.overwrite:
|
||||
if conflicts > 0 and not opts.overwrite:
|
||||
_error_block(
|
||||
f"Plan has {preview_conflicts} conflict(s). Refusing to apply.",
|
||||
f"Plan has {conflicts} conflict(s). Refusing to apply.",
|
||||
"Each conflict is an item whose target already exists in ~/.hermes/. "
|
||||
"Re-run with --overwrite to replace conflicting targets (item-level "
|
||||
"backups are written to the migration report directory).",
|
||||
"Or re-run with --dry-run to review the full plan.",
|
||||
)
|
||||
"Or re-run with --dry-run to review the full plan.")
|
||||
return False
|
||||
return True
|
||||
|
||||
|
||||
def _apply_migration(run_migrator: Callable[[bool], dict], opts: SimpleNamespace) -> None:
|
||||
"""Take a pre-migration backup (unless --no-backup), execute, and print the report.
|
||||
|
||||
The backup shares implementation with the pre-update backup (same exclusions, SQLite
|
||||
safe-copy, zip format) so it is restorable with `hermes import` — one atomic restore point
|
||||
before any mutation, auto-pruned to the last 5 pre-migration zips.
|
||||
"""
|
||||
"""Take a pre-migration backup (unless --no-backup), execute, and print the report. The backup
|
||||
shares the pre-update backup's implementation (exclusions, SQLite safe-copy, zip) so it is
|
||||
restorable with `hermes import`: one restore point before any mutation, pruned to the last 5."""
|
||||
backup_archive: Optional[Path] = None
|
||||
if not opts.no_backup:
|
||||
try:
|
||||
from hermes_cli.backup import create_pre_migration_backup, _format_size
|
||||
backup_archive = create_pre_migration_backup(hermes_home=opts.hermes_home)
|
||||
if backup_archive:
|
||||
size_str = _format_size(backup_archive.stat().st_size)
|
||||
print()
|
||||
print_success(f"Pre-migration backup: {backup_archive} ({size_str})")
|
||||
print_success(f"Pre-migration backup: {backup_archive} "
|
||||
f"({_format_size(backup_archive.stat().st_size)})")
|
||||
print_info(f"Restore with: hermes import {backup_archive.name}")
|
||||
except Exception as e:
|
||||
_error_block(
|
||||
return _error_block(
|
||||
f"Could not create pre-migration backup: {e}",
|
||||
"Re-run with --no-backup to skip, or free up disk space under the Hermes home.",
|
||||
)
|
||||
logger.debug("Pre-migration backup error", exc_info=True)
|
||||
return
|
||||
debug="Pre-migration backup error")
|
||||
try:
|
||||
report = run_migrator(True)
|
||||
except Exception as e:
|
||||
_error_block(f"Migration failed: {e}")
|
||||
logger.debug("OpenClaw migration error", exc_info=True)
|
||||
_error_block(f"Migration failed: {e}", debug="OpenClaw migration error")
|
||||
if backup_archive:
|
||||
print_info(f"A pre-migration backup is available at: {backup_archive}")
|
||||
print_info(f"Restore with: hermes import {backup_archive.name}")
|
||||
_info(f"A pre-migration backup is available at: {backup_archive}",
|
||||
f"Restore with: hermes import {backup_archive.name}")
|
||||
return
|
||||
_print_migration_report(report, dry_run=False)
|
||||
|
||||
|
||||
def _cmd_cleanup(args):
|
||||
"""Offer to rename leftover OpenClaw directories to .pre-migration to free disk space."""
|
||||
dry_run = getattr(args, "dry_run", False)
|
||||
auto_yes = getattr(args, "yes", False)
|
||||
explicit_source = getattr(args, "source", None)
|
||||
|
||||
dry_run, auto_yes = getattr(args, "dry_run", False), getattr(args, "yes", False)
|
||||
_print_banner("OpenClaw Cleanup")
|
||||
dirs_to_check = [Path(explicit_source)] if explicit_source else _find_openclaw_dirs()
|
||||
source = getattr(args, "source", None)
|
||||
dirs_to_check = [Path(source)] if source else _find_openclaw_dirs()
|
||||
if not dirs_to_check:
|
||||
print()
|
||||
print_success("No OpenClaw directories found. Nothing to clean up.")
|
||||
return
|
||||
|
||||
return print_success("No OpenClaw directories found. Nothing to clean up.")
|
||||
# Archiving while the service is active makes it recreate an empty skeleton directory.
|
||||
running = _detect_openclaw_processes()
|
||||
if running:
|
||||
_error_block(
|
||||
"OpenClaw appears to be still running:",
|
||||
*(f" * {detail}" for detail in running),
|
||||
"Archiving .openclaw/ while the service is active may cause it to "
|
||||
"immediately recreate an empty skeleton directory, destroying your config.",
|
||||
"Stop OpenClaw first: systemctl --user stop openclaw-gateway.service",
|
||||
)
|
||||
print()
|
||||
if not _confirm(
|
||||
auto_yes, "Proceed anyway?", default=False,
|
||||
non_tty=("Non-interactive session — aborting. Stop OpenClaw and re-run.",),
|
||||
declined="Aborted. Stop OpenClaw first, then re-run: hermes claw cleanup",
|
||||
):
|
||||
return
|
||||
|
||||
if running and not _warn_running(
|
||||
auto_yes, "OpenClaw appears to be still running:", running,
|
||||
("Archiving .openclaw/ while the service is active may cause it to "
|
||||
"immediately recreate an empty skeleton directory, destroying your config.",
|
||||
"Stop OpenClaw first: systemctl --user stop openclaw-gateway.service"),
|
||||
"Proceed anyway?",
|
||||
declined="Aborted. Stop OpenClaw first, then re-run: hermes claw cleanup",
|
||||
non_tty=("Non-interactive session — aborting. Stop OpenClaw and re-run.",),
|
||||
):
|
||||
return
|
||||
total_archived = 0
|
||||
for source_dir in dirs_to_check:
|
||||
_describe_openclaw_dir(source_dir)
|
||||
@@ -459,8 +411,8 @@ def _cmd_cleanup(args):
|
||||
n = len(dirs_to_check) if dry_run else total_archived
|
||||
word = "directory" if n == 1 else "directories"
|
||||
if dry_run:
|
||||
print_info(f"Dry run complete. {n} {word} would be archived.")
|
||||
print_info("Run without --dry-run to archive them.")
|
||||
_info(f"Dry run complete. {n} {word} would be archived.",
|
||||
"Run without --dry-run to archive them.")
|
||||
elif total_archived:
|
||||
print_success(f"Cleaned up {total_archived} OpenClaw {word}.")
|
||||
print_info("Directories were renamed, not deleted. You can undo by renaming them back.")
|
||||
@@ -474,17 +426,15 @@ def _describe_openclaw_dir(source_dir: Path) -> None:
|
||||
print_header(f"Found: {source_dir}")
|
||||
state_files = _scan_workspace_state(source_dir)
|
||||
try:
|
||||
workspace_dirs = [
|
||||
d for d in source_dir.iterdir()
|
||||
if d.is_dir() and not d.name.startswith(".")
|
||||
and any((d / n).exists() for n in ("todo.json", "SOUL.md", "MEMORY.md", "USER.md"))
|
||||
]
|
||||
workspace_dirs = [d for d in source_dir.iterdir() if d.is_dir()
|
||||
and not d.name.startswith(".")
|
||||
and any((d / n).exists() for n in _WORKSPACE_MARKERS)]
|
||||
except OSError:
|
||||
workspace_dirs = []
|
||||
|
||||
if workspace_dirs:
|
||||
print_info(f"Workspace directories: {len(workspace_dirs)}")
|
||||
_print_rows([_workspace_row(ws) for ws in workspace_dirs], 5)
|
||||
_print_rows([f"{ws.name}/ ({', '.join(_workspace_items(ws)) or 'empty'})"
|
||||
for ws in workspace_dirs], 5)
|
||||
if state_files:
|
||||
print()
|
||||
print(color(f" {len(state_files)} state file(s) found:", Colors.YELLOW))
|
||||
@@ -492,9 +442,8 @@ def _describe_openclaw_dir(source_dir: Path) -> None:
|
||||
print()
|
||||
|
||||
|
||||
def _workspace_row(ws: Path) -> str:
|
||||
items = [label for name, label, check in _WORKSPACE_ITEM_LABELS if check(ws / name)]
|
||||
return f"{ws.name}/ ({', '.join(items) or 'empty'})"
|
||||
def _workspace_items(ws: Path) -> list[str]:
|
||||
return [lbl for n, lbl, chk in _WORKSPACE_ITEM_LABELS if chk(ws / n)]
|
||||
|
||||
|
||||
def _print_rows(rows: list[str], shown: int) -> None:
|
||||
@@ -507,44 +456,37 @@ def _print_rows(rows: list[str], shown: int) -> None:
|
||||
|
||||
def _print_migration_report(report: dict, dry_run: bool):
|
||||
"""Print a formatted migration report."""
|
||||
summary = report.get("summary", {})
|
||||
summary, items = report.get("summary", {}), report.get("items", [])
|
||||
migrated = summary.get("migrated", 0)
|
||||
print()
|
||||
print_header("Dry Run Results" if dry_run else "Migration Results")
|
||||
if dry_run:
|
||||
print_info("No files were modified. This is a preview of what would happen.")
|
||||
print()
|
||||
|
||||
items = report.get("items", [])
|
||||
migrated_items = [i for i in items if i.get("status") == "migrated"]
|
||||
if migrated_items:
|
||||
print(color(f" ✓ {'Would migrate' if dry_run else 'Migrated'}:", Colors.GREEN))
|
||||
for item in migrated_items:
|
||||
migrated_heading = f" ✓ {'Would migrate' if dry_run else 'Migrated'}:"
|
||||
for status, heading, heading_color, default_reason in (
|
||||
("migrated", migrated_heading, Colors.GREEN, None), *_REPORT_REASON_GROUPS):
|
||||
rows = []
|
||||
for item in (i for i in items if i.get("status") == status):
|
||||
kind, dest = item.get("kind", "unknown"), item.get("destination", "")
|
||||
print(f" {kind:<22s} → {str(dest).replace(str(Path.home()), '~')}" if dest
|
||||
else f" {kind}")
|
||||
print()
|
||||
for status, heading, heading_color, default_reason in _REPORT_REASON_GROUPS:
|
||||
group = [i for i in items if i.get("status") == status]
|
||||
if not group:
|
||||
continue
|
||||
print(color(heading, heading_color))
|
||||
for item in group:
|
||||
print(f" {item.get('kind', 'unknown'):<22s} {item.get('reason', default_reason)}")
|
||||
print()
|
||||
|
||||
counts = ((migrated, "would migrate" if dry_run else "migrated"),
|
||||
(summary.get("conflict", 0), "conflict(s)"), (summary.get("skipped", 0), "skipped"),
|
||||
(summary.get("error", 0), "error(s)"))
|
||||
if default_reason is not None: # conflict / skipped / error rows show a reason
|
||||
rows.append(f"{kind:<22s} {item.get('reason', default_reason)}")
|
||||
else:
|
||||
rows.append(f"{kind:<22s} → {str(dest).replace(str(Path.home()), '~')}" if dest
|
||||
else kind)
|
||||
if rows:
|
||||
print(color(heading, heading_color))
|
||||
_print_rows(rows, len(rows))
|
||||
print()
|
||||
counts = [(migrated, "would migrate" if dry_run else "migrated")] + [
|
||||
(summary.get(k, 0), label) for k, label in _SUMMARY_COUNT_LABELS]
|
||||
parts = [f"{count} {label}" for count, label in counts if count]
|
||||
print_info(f"Summary: {', '.join(parts)}" if parts else "Nothing to migrate.")
|
||||
if report.get("output_dir"):
|
||||
print_info(f"Full report saved to: {report['output_dir']}")
|
||||
|
||||
if dry_run:
|
||||
print()
|
||||
print_info("To execute the migration, run without --dry-run:")
|
||||
print_info(f" hermes claw migrate --preset {report.get('preset', 'full')}")
|
||||
_info("", "To execute the migration, run without --dry-run:",
|
||||
f" hermes claw migrate --preset {report.get('preset', 'full')}")
|
||||
elif migrated:
|
||||
print()
|
||||
print_success("Migration complete!")
|
||||
@@ -556,9 +498,6 @@ def _print_migration_report(report: dict, dry_run: bool):
|
||||
" Your OPENROUTER_API_KEY and other provider keys must be added manually.",
|
||||
):
|
||||
print(color(line, Colors.YELLOW))
|
||||
print()
|
||||
print_info("To migrate API keys, re-run with:")
|
||||
print_info(" hermes claw migrate --migrate-secrets")
|
||||
print()
|
||||
print_info("Or add your key manually:")
|
||||
print_info(" hermes config set OPENROUTER_API_KEY sk-or-v1-...")
|
||||
_info("", "To migrate API keys, re-run with:",
|
||||
" hermes claw migrate --migrate-secrets", "", "Or add your key manually:",
|
||||
" hermes config set OPENROUTER_API_KEY sk-or-v1-...")
|
||||
|
||||
Reference in New Issue
Block a user