From 67be468ff126002812f333ac127e9344d5233dbe Mon Sep 17 00:00:00 2001 From: AaronZ345 <34849476+AaronZ345@users.noreply.github.com> Date: Fri, 17 Jul 2026 11:58:44 +0800 Subject: [PATCH] feat(cli): add portable data backup and restore --- cli/cli.py | 5 + cli/commands/backup.py | 332 +++++++++++++++++++++++++++++++++++++++ docs/cli/backup.mdx | 51 ++++++ docs/cli/index.mdx | 5 + docs/docs.json | 2 + docs/zh/cli/backup.mdx | 45 ++++++ docs/zh/cli/index.mdx | 5 + tests/test_cli_backup.py | 110 +++++++++++++ 8 files changed, 555 insertions(+) create mode 100644 cli/commands/backup.py create mode 100644 docs/cli/backup.mdx create mode 100644 docs/zh/cli/backup.mdx create mode 100644 tests/test_cli_backup.py diff --git a/cli/cli.py b/cli/cli.py index b266b372..01e70cf8 100644 --- a/cli/cli.py +++ b/cli/cli.py @@ -7,6 +7,7 @@ from cli.commands.process import start, stop, restart, self_restart, update, sta from cli.commands.context import context from cli.commands.install import install_browser from cli.commands.knowledge import knowledge +from cli.commands.backup import backup_command, restore_command HELP_TEXT = """Usage: cow COMMAND [ARGS]... @@ -24,6 +25,8 @@ Commands: logs View CowAgent logs. skill Manage CowAgent skills. knowledge Manage knowledge base. + backup Back up config and agent workspace. + restore Restore a CowAgent backup. install-browser Install browser tool (Playwright + Chromium). Tip: Memory index management lives in chat — send /memory status or @@ -74,6 +77,8 @@ main.add_command(status) main.add_command(logs) main.add_command(context) main.add_command(knowledge) +main.add_command(backup_command) +main.add_command(restore_command) main.add_command(install_browser) diff --git a/cli/commands/backup.py b/cli/commands/backup.py new file mode 100644 index 00000000..60360c81 --- /dev/null +++ b/cli/commands/backup.py @@ -0,0 +1,332 @@ +"""Portable local backup and restore commands for CowAgent user data.""" + +import json +import os +import shutil +import stat +import tempfile +import zipfile +from datetime import datetime, timezone +from pathlib import Path, PurePosixPath +from typing import Iterable, Optional, Set + +import click + +from cli.utils import get_project_root + + +BACKUP_FORMAT = "cowagent-backup" +BACKUP_VERSION = 1 +_SKIP_DIRS = {".git", "__pycache__", "tmp"} +_SKIP_FILES = {".DS_Store"} + + +def _data_root() -> Path: + configured = os.environ.get("COW_DATA_DIR") + return Path(configured).expanduser().resolve() if configured else Path(get_project_root()).resolve() + + +def _read_config(data_root: Path) -> dict: + path = data_root / "config.json" + if not path.is_file(): + return {} + try: + with path.open("r", encoding="utf-8") as handle: + value = json.load(handle) + return value if isinstance(value, dict) else {} + except (OSError, ValueError): + return {} + + +def _workspace_from_config(config: dict) -> Path: + return Path(config.get("agent_workspace") or "~/cow").expanduser().resolve() + + +def _legacy_user_data_path(data_root: Path, config: dict) -> Path: + appdata_dir = config.get("appdata_dir") or "" + return (data_root / appdata_dir / "user_datas.pkl").resolve() + + +def _is_within(path: Path, root: Path) -> bool: + try: + return os.path.commonpath([str(path.resolve()), str(root.resolve())]) == str(root.resolve()) + except (OSError, ValueError): + return False + + +def _iter_workspace_files(workspace: Path, excluded: Set[Path]): + if not workspace.is_dir(): + return + for current, dirnames, filenames in os.walk(str(workspace), followlinks=False): + current_path = Path(current) + dirnames[:] = [ + name for name in dirnames + if name not in _SKIP_DIRS and not (current_path / name).is_symlink() + ] + for name in filenames: + path = current_path / name + if name in _SKIP_FILES or name.endswith((".pyc", ".pyo")): + continue + if path.is_symlink() or path.resolve() in excluded: + continue + if path.is_file(): + yield path + + +def create_backup_archive( + output: Path, + data_root: Path, + workspace: Path, + excluded_paths: Optional[Iterable[Path]] = None, +) -> dict: + """Create a portable archive containing config and the agent workspace.""" + output = Path(output).expanduser().resolve() + data_root = Path(data_root).expanduser().resolve() + workspace = Path(workspace).expanduser().resolve() + output.parent.mkdir(parents=True, exist_ok=True) + excluded = {Path(path).expanduser().resolve() for path in (excluded_paths or [])} + excluded.add(output) + + config_path = data_root / "config.json" + config = _read_config(data_root) + legacy_path = _legacy_user_data_path(data_root, config) + workspace_files = list(_iter_workspace_files(workspace, excluded)) + total_bytes = sum(path.stat().st_size for path in workspace_files) + + manifest = { + "format": BACKUP_FORMAT, + "version": BACKUP_VERSION, + "created_at": datetime.now(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z"), + "workspace_source": str(workspace), + "contents": { + "config": config_path.is_file(), + "legacy_user_data": legacy_path.is_file(), + "workspace_files": len(workspace_files), + "workspace_bytes": total_bytes, + }, + } + + temp_dir = Path(tempfile.mkdtemp(prefix="cowagent-backup-")) + temp_archive = temp_dir / "backup.zip" + try: + with zipfile.ZipFile( + str(temp_archive), "w", compression=zipfile.ZIP_DEFLATED, allowZip64=True + ) as archive: + archive.writestr( + "manifest.json", + json.dumps(manifest, ensure_ascii=False, indent=2) + "\n", + ) + if config_path.is_file(): + archive.write(str(config_path), "data/config.json") + if legacy_path.is_file(): + archive.write(str(legacy_path), "data/user_datas.pkl") + for path in workspace_files: + relative = path.relative_to(workspace).as_posix() + archive.write(str(path), "workspace/" + relative) + os.replace(str(temp_archive), str(output)) + try: + os.chmod(str(output), stat.S_IRUSR | stat.S_IWUSR) + except OSError: + pass + finally: + shutil.rmtree(str(temp_dir), ignore_errors=True) + + manifest["archive"] = str(output) + return manifest + + +def _validate_archive(archive: zipfile.ZipFile) -> dict: + names = {info.filename for info in archive.infolist()} + if "manifest.json" not in names: + raise ValueError("archive is missing manifest.json") + try: + manifest = json.loads(archive.read("manifest.json").decode("utf-8")) + except (ValueError, UnicodeDecodeError) as exc: + raise ValueError("archive manifest is invalid") from exc + if manifest.get("format") != BACKUP_FORMAT or manifest.get("version") != BACKUP_VERSION: + raise ValueError("unsupported CowAgent backup format or version") + + for info in archive.infolist(): + name = info.filename + path = PurePosixPath(name) + if not name or path.is_absolute() or ".." in path.parts or "\\" in name: + raise ValueError(f"unsafe archive path: {name!r}") + mode = (info.external_attr >> 16) & 0o170000 + if mode == stat.S_IFLNK: + raise ValueError(f"symbolic links are not allowed in backups: {name!r}") + if name != "manifest.json" and not name.startswith(("data/", "workspace/")): + raise ValueError(f"unexpected archive entry: {name!r}") + return manifest + + +def _extract_validated(archive: zipfile.ZipFile, destination: Path) -> None: + for info in archive.infolist(): + target = destination.joinpath(*PurePosixPath(info.filename).parts) + if info.is_dir(): + target.mkdir(parents=True, exist_ok=True) + continue + target.parent.mkdir(parents=True, exist_ok=True) + with archive.open(info, "r") as source, target.open("wb") as output: + shutil.copyfileobj(source, output) + + +def _atomic_copy(source: Path, destination: Path, private: bool = False) -> None: + destination.parent.mkdir(parents=True, exist_ok=True) + fd, temp_name = tempfile.mkstemp(prefix=destination.name + ".", dir=str(destination.parent)) + os.close(fd) + try: + shutil.copy2(str(source), temp_name) + os.replace(temp_name, str(destination)) + if private: + try: + os.chmod(str(destination), stat.S_IRUSR | stat.S_IWUSR) + except OSError: + pass + finally: + if os.path.exists(temp_name): + os.remove(temp_name) + + +def restore_backup_archive( + archive_path: Path, + data_root: Path, + workspace: Optional[Path] = None, +) -> dict: + """Merge a validated backup into the selected data root and workspace.""" + archive_path = Path(archive_path).expanduser().resolve() + data_root = Path(data_root).expanduser().resolve() + current_config = _read_config(data_root) + + temp_dir = Path(tempfile.mkdtemp(prefix="cowagent-restore-")) + try: + with zipfile.ZipFile(str(archive_path), "r") as archive: + manifest = _validate_archive(archive) + _extract_validated(archive, temp_dir) + + archived_config_path = temp_dir / "data" / "config.json" + archived_config = {} + if archived_config_path.is_file(): + with archived_config_path.open("r", encoding="utf-8") as handle: + value = json.load(handle) + if not isinstance(value, dict): + raise ValueError("archived config.json must contain an object") + archived_config = value + + if workspace is not None: + target_workspace = Path(workspace).expanduser().resolve() + elif current_config.get("agent_workspace"): + target_workspace = _workspace_from_config(current_config) + else: + # Do not trust an archive-controlled absolute destination on a + # fresh machine. Portable restores default to the standard local + # workspace unless the operator supplies --workspace. + target_workspace = Path("~/cow").expanduser().resolve() + + restored_config = dict(archived_config) + if restored_config: + restored_config["agent_workspace"] = str(target_workspace) + appdata_dir = restored_config.get("appdata_dir") or "" + if appdata_dir: + archived_appdata = (data_root / appdata_dir).resolve() + if not _is_within(archived_appdata, data_root): + # Keep legacy user data under the selected data root + # instead of writing to an archive-controlled path. + restored_config["appdata_dir"] = "" + config_temp = temp_dir / "restored-config.json" + with config_temp.open("w", encoding="utf-8") as handle: + json.dump(restored_config, handle, ensure_ascii=False, indent=2) + handle.write("\n") + _atomic_copy(config_temp, data_root / "config.json", private=True) + + workspace_root = temp_dir / "workspace" + restored_files = 0 + if workspace_root.is_dir(): + for source in _iter_workspace_files(workspace_root, set()): + relative = source.relative_to(workspace_root) + destination = target_workspace / relative + if not _is_within(destination, target_workspace): + raise ValueError(f"unsafe workspace destination: {relative}") + _atomic_copy(source, destination) + restored_files += 1 + + legacy_source = temp_dir / "data" / "user_datas.pkl" + if legacy_source.is_file(): + effective_config = restored_config or current_config + legacy_destination = _legacy_user_data_path(data_root, effective_config) + _atomic_copy(legacy_source, legacy_destination, private=True) + + return { + "manifest": manifest, + "workspace": str(target_workspace), + "workspace_files": restored_files, + "config_restored": bool(restored_config), + "legacy_user_data_restored": legacy_source.is_file(), + } + finally: + shutil.rmtree(str(temp_dir), ignore_errors=True) + + +@click.command("backup") +@click.option( + "--output", + "-o", + type=click.Path(dir_okay=False, path_type=Path), + help="Output .zip path (default: ./cow-backup-.zip).", +) +def backup_command(output: Optional[Path]): + """Back up config, persona, memory, skills, knowledge, and schedules.""" + data_root = _data_root() + config = _read_config(data_root) + workspace = _workspace_from_config(config) + if output is None: + stamp = datetime.now().strftime("%Y%m%d-%H%M%S") + output = Path.cwd() / f"cow-backup-{stamp}.zip" + result = create_backup_archive(output, data_root, workspace) + click.echo(click.style("✓ Backup created", fg="green")) + click.echo(f" Archive: {result['archive']}") + click.echo(f" Workspace files: {result['contents']['workspace_files']}") + click.echo(" Keep this archive private: it may contain API keys and personal data.") + + +@click.command("restore") +@click.argument("archive", type=click.Path(exists=True, dir_okay=False, path_type=Path)) +@click.option( + "--workspace", + type=click.Path(file_okay=False, path_type=Path), + help="Restore workspace files to this directory.", +) +@click.option("--yes", is_flag=True, help="Confirm overwriting matching files.") +def restore_command(archive: Path, workspace: Optional[Path], yes: bool): + """Restore a backup without deleting unrelated destination files.""" + from cli.commands.process import _read_pid + + pid = _read_pid() + if pid: + raise click.ClickException( + f"CowAgent is running (PID: {pid}). Run 'cow stop' before restoring." + ) + if not yes: + click.confirm( + "Restore this archive and overwrite matching config/workspace files?", + abort=True, + ) + + data_root = _data_root() + current_config = _read_config(data_root) + current_workspace = _workspace_from_config(current_config) + has_current_data = (data_root / "config.json").is_file() or current_workspace.is_dir() + if has_current_data: + stamp = datetime.now().strftime("%Y%m%d-%H%M%S") + rollback = archive.resolve().parent / f"cow-pre-restore-{stamp}.zip" + create_backup_archive( + rollback, + data_root, + current_workspace, + excluded_paths={archive.resolve()}, + ) + click.echo(f"Rollback backup: {rollback}") + + result = restore_backup_archive(archive, data_root, workspace) + click.echo(click.style("✓ Backup restored", fg="green")) + click.echo(f" Workspace: {result['workspace']}") + click.echo(f" Restored files: {result['workspace_files']}") diff --git a/docs/cli/backup.mdx b/docs/cli/backup.mdx new file mode 100644 index 00000000..d4eddca7 --- /dev/null +++ b/docs/cli/backup.mdx @@ -0,0 +1,51 @@ +--- +title: Backup and Restore +description: Export and restore CowAgent configuration and agent workspace data +--- + +CowAgent can create a portable local archive for migration or disaster recovery. + +## Create a backup + +```bash +cow backup +cow backup --output /safe/location/cow-backup.zip +``` + +The archive contains: + +- `config.json` +- Agent persona and user files such as `AGENT.md`, `USER.md`, and `MEMORY.md` +- Daily memory, session history, knowledge, custom skills, and scheduled tasks in the configured workspace +- Legacy `user_datas.pkl`, when present + +Transient `tmp/` data, caches, Git metadata, and symbolic links are skipped. The +archive is written with owner-only permissions where the operating system +supports them. + + + A backup may contain API keys, conversation history, and other personal data. + Store and transfer it as a secret. The ZIP file is not encrypted. + + +## Restore a backup + +Stop CowAgent before restoring: + +```bash +cow stop +cow restore /safe/location/cow-backup.zip +``` + +Use `--workspace` to migrate workspace files to a different location: + +```bash +cow restore cow-backup.zip --workspace ~/cow-restored +``` + +Restore validates the archive format and paths before writing anything. It +overwrites matching files but does not delete unrelated files already present +at the destination. When current CowAgent data exists, the command first +creates a `cow-pre-restore-*.zip` rollback archive beside the selected backup. + +For unattended scripts, pass `--yes` to acknowledge overwrites. diff --git a/docs/cli/index.mdx b/docs/cli/index.mdx index e13b45a3..10f090b4 100644 --- a/docs/cli/index.mdx +++ b/docs/cli/index.mdx @@ -44,6 +44,10 @@ Memory & Knowledge: memory Memory distillation (dream) knowledge View knowledge base stats and structure +Data portability: + backup Back up config and agent workspace + restore Restore a CowAgent backup + Others: help Show this help message version Show version @@ -90,6 +94,7 @@ In the Web console or any connected channel, type `/` to see command suggestions | start / stop / restart | ✓ | ✗ | | update | ✓ | ✗ | | install-browser | ✓ | ✗ | +| backup / restore | ✓ | ✗ | `context` only shows a hint in the terminal to use it in chat. `config` is only available in chat. diff --git a/docs/docs.json b/docs/docs.json index 66a37696..f4b07278 100644 --- a/docs/docs.json +++ b/docs/docs.json @@ -246,6 +246,7 @@ "cli/process", "cli/skill", "cli/memory-knowledge", + "cli/backup", "cli/general" ] } @@ -473,6 +474,7 @@ "zh/cli/process", "zh/cli/skill", "zh/cli/memory-knowledge", + "zh/cli/backup", "zh/cli/general" ] } diff --git a/docs/zh/cli/backup.mdx b/docs/zh/cli/backup.mdx new file mode 100644 index 00000000..cc9b1166 --- /dev/null +++ b/docs/zh/cli/backup.mdx @@ -0,0 +1,45 @@ +--- +title: 备份与恢复 +description: 导出和恢复 CowAgent 配置及 Agent 工作区数据 +--- + +CowAgent 可以生成便携的本地归档，用于设备迁移或灾难恢复。 + +## 创建备份 + +```bash +cow backup +cow backup --output /safe/location/cow-backup.zip +``` + +归档包含： + +- `config.json` +- `AGENT.md`、`USER.md`、`MEMORY.md` 等 Agent 人格和用户文件 +- 工作区内的每日记忆、会话历史、知识库、自定义技能和定时任务 +- 旧版 `user_datas.pkl`（如存在） + +临时 `tmp/` 数据、缓存、Git 元数据和符号链接不会写入归档。在操作系统支持的情况下，归档文件权限会限制为仅当前用户可读写。 + + + 备份中可能含有 API 密钥、对话历史及其他个人数据，应按密钥文件保存和传输。ZIP 文件本身没有加密。 + + +## 恢复备份 + +恢复前先停止 CowAgent： + +```bash +cow stop +cow restore /safe/location/cow-backup.zip +``` + +迁移到不同工作区时，可以显式指定路径： + +```bash +cow restore cow-backup.zip --workspace ~/cow-restored +``` + +恢复命令会在写入前校验归档格式和所有路径。它会覆盖同名文件，但不会删除目标目录中无关的现有文件。如果检测到当前 CowAgent 数据，会先在所选归档旁创建 `cow-pre-restore-*.zip` 回滚备份。 + +无人值守脚本可传入 `--yes` 确认覆盖。 diff --git a/docs/zh/cli/index.mdx b/docs/zh/cli/index.mdx index f6462ecb..7ed48297 100644 --- a/docs/zh/cli/index.mdx +++ b/docs/zh/cli/index.mdx @@ -44,6 +44,10 @@ Memory & Knowledge: memory Memory distillation (dream) knowledge View knowledge base stats and structure +Data portability: + backup Back up config and agent workspace + restore Restore a CowAgent backup + Others: help Show this help message version Show version @@ -92,6 +96,7 @@ Others: | start / stop / restart | ✓ | ✗ | | update | ✓ | ✗ | | install-browser | ✓ | ✗ | +| backup / restore | ✓ | ✗ | `context` 在终端中仅提示到对话中使用。`config` 仅支持在对话中修改。 diff --git a/tests/test_cli_backup.py b/tests/test_cli_backup.py new file mode 100644 index 00000000..73db2e11 --- /dev/null +++ b/tests/test_cli_backup.py @@ -0,0 +1,110 @@ +"""Tests for portable CowAgent backup archives.""" + +import json +import zipfile +from pathlib import Path + +import pytest + +from cli.commands.backup import create_backup_archive, restore_backup_archive + + +def _write_json(path: Path, value: dict): + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(json.dumps(value), encoding="utf-8") + + +def test_backup_restore_round_trip(tmp_path): + source_data = tmp_path / "source-data" + source_workspace = tmp_path / "source-workspace" + _write_json(source_data / "config.json", { + "agent_workspace": str(source_workspace), + "open_ai_api_key": "secret-value", + }) + (source_workspace / "memory").mkdir(parents=True) + (source_workspace / "scheduler").mkdir() + (source_workspace / "knowledge").mkdir() + (source_workspace / "USER.md").write_text("# User\n", encoding="utf-8") + (source_workspace / "MEMORY.md").write_text("remember this\n", encoding="utf-8") + (source_workspace / "memory" / "2026-07-17.md").write_text("daily\n", encoding="utf-8") + (source_workspace / "scheduler" / "tasks.json").write_text("{}\n", encoding="utf-8") + (source_workspace / "knowledge" / "index.md").write_text("# Index\n", encoding="utf-8") + (source_workspace / "tmp").mkdir() + (source_workspace / "tmp" / "scratch.txt").write_text("skip", encoding="utf-8") + + archive = tmp_path / "cow-backup.zip" + summary = create_backup_archive(archive, source_data, source_workspace) + assert summary["contents"]["workspace_files"] == 5 + assert archive.exists() + + target_data = tmp_path / "target-data" + target_workspace = tmp_path / "target-workspace" + result = restore_backup_archive(archive, target_data, target_workspace) + + restored_config = json.loads((target_data / "config.json").read_text(encoding="utf-8")) + assert restored_config["agent_workspace"] == str(target_workspace.resolve()) + assert restored_config["open_ai_api_key"] == "secret-value" + assert (target_workspace / "MEMORY.md").read_text(encoding="utf-8") == "remember this\n" + assert (target_workspace / "scheduler" / "tasks.json").exists() + assert (target_workspace / "knowledge" / "index.md").exists() + assert not (target_workspace / "tmp" / "scratch.txt").exists() + assert result["workspace_files"] == 5 + + +def test_restore_merges_without_deleting_unrelated_files(tmp_path): + source_data = tmp_path / "source-data" + source_workspace = tmp_path / "source-workspace" + _write_json(source_data / "config.json", {"agent_workspace": str(source_workspace)}) + source_workspace.mkdir() + (source_workspace / "MEMORY.md").write_text("new\n", encoding="utf-8") + archive = tmp_path / "cow-backup.zip" + create_backup_archive(archive, source_data, source_workspace) + + target_workspace = tmp_path / "target-workspace" + target_workspace.mkdir() + (target_workspace / "MEMORY.md").write_text("old\n", encoding="utf-8") + (target_workspace / "keep.txt").write_text("keep\n", encoding="utf-8") + + restore_backup_archive(archive, tmp_path / "target-data", target_workspace) + assert (target_workspace / "MEMORY.md").read_text(encoding="utf-8") == "new\n" + assert (target_workspace / "keep.txt").read_text(encoding="utf-8") == "keep\n" + + +def test_restore_rejects_path_traversal(tmp_path): + archive = tmp_path / "malicious.zip" + manifest = {"format": "cowagent-backup", "version": 1} + with zipfile.ZipFile(str(archive), "w") as output: + output.writestr("manifest.json", json.dumps(manifest)) + output.writestr("workspace/../../escape.txt", "nope") + + with pytest.raises(ValueError, match="unsafe archive path"): + restore_backup_archive(archive, tmp_path / "data", tmp_path / "workspace") + + +def test_fresh_restore_ignores_archive_controlled_destinations(tmp_path, monkeypatch): + source_data = tmp_path / "source-data" + source_workspace = tmp_path / "source-workspace" + outside_appdata = tmp_path / "outside-appdata" + archive_workspace = tmp_path / "archive-controlled-workspace" + _write_json(source_data / "config.json", { + "agent_workspace": str(archive_workspace), + "appdata_dir": str(outside_appdata), + }) + source_workspace.mkdir() + (source_workspace / "MEMORY.md").write_text("portable\n", encoding="utf-8") + outside_appdata.mkdir() + (outside_appdata / "user_datas.pkl").write_bytes(b"legacy") + archive = tmp_path / "cow-backup.zip" + create_backup_archive(archive, source_data, source_workspace) + + fake_home = tmp_path / "home" + monkeypatch.setenv("HOME", str(fake_home)) + target_data = tmp_path / "target-data" + result = restore_backup_archive(archive, target_data) + + assert result["workspace"] == str((fake_home / "cow").resolve()) + assert (fake_home / "cow" / "MEMORY.md").exists() + assert not archive_workspace.exists() + assert (target_data / "user_datas.pkl").read_bytes() == b"legacy" + restored_config = json.loads((target_data / "config.json").read_text(encoding="utf-8")) + assert restored_config["appdata_dir"] == ""