cursor-chat-archive/scripts/restore-archive.py

#!/usr/bin/env python3
"""Restore Cursor chat history from cursor-chat-archive on a new machine/path."""

from __future__ import annotations

import hashlib
import json
import shutil
import sqlite3
import subprocess
import sys
from pathlib import Path
from urllib.parse import unquote, urlparse

ARCHIVE_ROOT = Path(__file__).resolve().parents[1]
CONFIG_PATH = ARCHIVE_ROOT / "restore-config.json"
MANIFEST_PATH = ARCHIVE_ROOT / "manifest.json"
SOURCE_PROFILE_PATH = ARCHIVE_ROOT / "source-profile.json"


def folder_uri(path: Path) -> str:
    return f"file://{path.resolve()}"


def workspace_id(path: Path) -> str:
    return hashlib.md5(folder_uri(path).encode()).hexdigest()


def path_to_slug(path: Path) -> str:
    return str(path.resolve()).strip("/").replace("/", "-")


def load_config() -> dict:
    if not CONFIG_PATH.is_file():
        print(f"Create {CONFIG_PATH} from restore-config.example.json", file=sys.stderr)
        sys.exit(1)
    return json.loads(CONFIG_PATH.read_text(encoding="utf-8"))


def cursor_is_running() -> bool:
    try:
        out = subprocess.run(
            ["pgrep", "-f", "cursor.mjs"],
            capture_output=True,
            text=True,
            check=False,
        )
        return bool(out.stdout.strip())
    except FileNotFoundError:
        return False


def make_workspace_identifier(ws_id: str, abs_path: Path) -> dict:
    p = str(abs_path.resolve())
    return {
        "id": ws_id,
        "uri": {
            "$mid": 1,
            "fsPath": p,
            "external": f"file://{p}",
            "path": p,
            "scheme": "file",
        },
    }


def replace_bytes(blob: bytes, replacements: list[tuple[bytes, bytes]]) -> bytes:
    out = blob
    for old, new in replacements:
        if old in out:
            out = out.replace(old, new)
    return out


def replace_in_json_obj(obj, old_path: str, new_path: str, old_ids: set[str], new_id: str) -> bool:
    changed = False
    if isinstance(obj, dict):
        if obj.get("id") in old_ids:
            obj["id"] = new_id
            changed = True
        uri = obj.get("uri")
        if isinstance(uri, dict):
            for key in ("fsPath", "path"):
                if uri.get(key) == old_path:
                    uri[key] = new_path
                    changed = True
            if uri.get("external") == f"file://{old_path}":
                uri["external"] = f"file://{new_path}"
                changed = True
        for v in obj.values():
            if replace_in_json_obj(v, old_path, new_path, old_ids, new_id):
                changed = True
    elif isinstance(obj, list):
        for item in obj:
            if replace_in_json_obj(item, old_path, new_path, old_ids, new_id):
                changed = True
    elif isinstance(obj, str):
        if obj == old_path:
            return False
    return changed


def remap_global_db(
    db_path: Path,
    old_home: str,
    new_home: str,
    repo_mappings: list[dict],
) -> dict:
    conn = sqlite3.connect(db_path)
    stats = {"item_updates": 0, "kv_updates": 0, "header_updates": 0}

    home_repl = [
        (old_home.encode(), new_home.encode()),
        (old_home.replace("/", "\\").encode(), new_home.replace("/", "\\").encode()),
    ]

    for old_path, new_path, old_ids, new_id, composer_ids in repo_mappings:
        old_b = old_path.encode()
        new_b = new_path.encode()
        repl = home_repl + [(old_b, new_b)]
        for old_ws in old_ids:
            repl.append((old_ws.encode(), new_id.encode()))

        # composer headers
        row = conn.execute(
            "SELECT value FROM ItemTable WHERE key='composer.composerHeaders'"
        ).fetchone()
        if row:
            data = json.loads(row[0])
            for composer in data.get("allComposers", []):
                cid = composer.get("composerId", "")
                wi = composer.get("workspaceIdentifier") or {}
                uri = wi.get("uri") or {}
                path = uri.get("fsPath") or uri.get("path") or ""
                ws = wi.get("id", "")
                if (
                    cid in composer_ids
                    or path == old_path
                    or ws in old_ids
                ):
                    composer["workspaceIdentifier"] = json.loads(
                        json.dumps(make_workspace_identifier(new_id, Path(new_path)))
                    )
                    stats["header_updates"] += 1
            conn.execute(
                "UPDATE ItemTable SET value=? WHERE key='composer.composerHeaders'",
                (json.dumps(data, ensure_ascii=False),),
            )

        for cid in composer_ids:
            key = f"composerData:{cid}"
            row = conn.execute(
                "SELECT value FROM cursorDiskKV WHERE key=?", (key,)
            ).fetchone()
            if not row:
                continue
            try:
                data = json.loads(row[0])
                if replace_in_json_obj(data, old_path, new_path, set(old_ids), new_id):
                    conn.execute(
                        "UPDATE cursorDiskKV SET value=? WHERE key=?",
                        (json.dumps(data, ensure_ascii=False), key),
                    )
                    stats["kv_updates"] += 1
            except json.JSONDecodeError:
                pass

    # broad replace in ItemTable and cursorDiskKV strings
    for table, col in (("ItemTable", "value"), ("cursorDiskKV", "value")):
        rows = conn.execute(f"SELECT rowid, {col} FROM {table}").fetchall()
        for rowid, value in rows:
            if value is None:
                continue
            if isinstance(value, str):
                new_val = value
                for old_path, new_path, old_ids, new_id, _ in repo_mappings:
                    new_val = new_val.replace(old_path, new_path)
                    for oid in old_ids:
                        new_val = new_val.replace(oid, new_id)
                new_val = new_val.replace(old_home, new_home)
                if new_val != value:
                    conn.execute(
                        f"UPDATE {table} SET {col}=? WHERE rowid=?",
                        (new_val, rowid),
                    )
                    stats["item_updates"] += 1
            elif isinstance(value, bytes):
                new_val = value
                for old, new in home_repl:
                    new_val = replace_bytes(new_val, [(old, new)])
                for old_path, new_path, old_ids, new_id, _ in repo_mappings:
                    new_val = replace_bytes(
                        new_val, [(old_path.encode(), new_path.encode())]
                    )
                    for oid in old_ids:
                        new_val = replace_bytes(
                            new_val, [(oid.encode(), new_id.encode())]
                        )
                if new_val != value:
                    conn.execute(
                        f"UPDATE {table} SET {col}=? WHERE rowid=?",
                        (new_val, rowid),
                    )
                    stats["item_updates"] += 1

    conn.commit()
    conn.close()
    return stats


def copy_tree(src: Path, dst: Path) -> None:
    dst.parent.mkdir(parents=True, exist_ok=True)
    if dst.exists():
        shutil.rmtree(dst)
    if src.exists():
        shutil.copytree(src, dst, dirs_exist_ok=True)


def fix_workspace_json(ws_dir: Path, new_uri: str) -> None:
    wj = ws_dir / "workspace.json"
    wj.write_text(
        json.dumps({"folder": new_uri}, ensure_ascii=False, indent=2) + "\n",
        encoding="utf-8",
    )


def main() -> None:
    if cursor_is_running():
        sys.exit("Cursor is running. Quit Cursor completely, then run again.")

    cfg = load_config()
    new_home = Path(cfg["new_home"]).resolve()
    new_repos_root = Path(cfg["new_repos_root"]).resolve()
    new_projects = new_home / ".cursor" / "projects"
    new_ws_root = new_home / ".config" / "Cursor" / "User" / "workspaceStorage"
    new_global = new_home / ".config" / "Cursor" / "User" / "globalStorage"

    manifest = json.loads(MANIFEST_PATH.read_text(encoding="utf-8"))
    source = json.loads(SOURCE_PROFILE_PATH.read_text(encoding="utf-8"))
    old_home = source["old_home"]
    repos = manifest["repos"]

    ts = __import__("datetime").datetime.now().strftime("%Y%m%d%H%M%S")
    backup_root = new_home / ".cursor" / f"restore-chat-backup-{ts}"
    backup_root.mkdir(parents=True, exist_ok=True)
    for p in (new_global / "state.vscdb", new_global / "storage.json"):
        if p.exists():
            shutil.copy2(p, backup_root / p.name)

    # global
    archived_db = ARCHIVE_ROOT / "global" / "state.vscdb"
    if archived_db.exists():
        new_global.mkdir(parents=True, exist_ok=True)
        shutil.copy2(archived_db, new_global / "state.vscdb")
    archived_storage = ARCHIVE_ROOT / "global" / "storage.json"
    if archived_storage.exists():
        text = archived_storage.read_text(encoding="utf-8")
        text = text.replace(old_home, str(new_home))
        (new_global / "storage.json").write_text(text, encoding="utf-8")

    repo_mappings = []
    restore_report = []

    for repo_name, meta in repos.items():
        old_path = meta["old_absolute_path"]
        new_path = new_repos_root / repo_name
        new_uri = folder_uri(new_path)
        new_id = workspace_id(new_path)
        new_slug = path_to_slug(new_path)
        old_ids = meta.get("old_workspace_ids") or []
        if meta.get("primary_workspace_id") and meta["primary_workspace_id"] not in old_ids:
            old_ids.append(meta["primary_workspace_id"])
        composer_ids = set(meta.get("composer_ids") or [])
        repo_mappings.append(
            (old_path, str(new_path), old_ids, new_id, composer_ids)
        )

        repo_archive = ARCHIVE_ROOT / "repos" / repo_name

        # projects / agent-transcripts
        at_src = repo_archive / "agent-transcripts"
        if at_src.exists():
            copy_tree(at_src, new_projects / new_slug / "agent-transcripts")
        for extra in repo_archive.glob("agent-transcripts-extra/*"):
            if extra.is_dir():
                copy_tree(extra, new_projects / extra.name / "agent-transcripts")

        # workspace storage — use primary id dir from archive
        primary = meta.get("primary_workspace_id")
        ws_src = None
        if primary:
            candidate = repo_archive / "workspace-storage" / primary
            if candidate.exists():
                ws_src = candidate
        if ws_src is None:
            ws_dirs = list((repo_archive / "workspace-storage").glob("*"))
            ws_src = ws_dirs[0] if ws_dirs else None
        if ws_src and ws_src.exists():
            dst_ws = new_ws_root / new_id
            copy_tree(ws_src, dst_ws)
            fix_workspace_json(dst_ws, new_uri)

        restore_report.append(
            {
                "repo": repo_name,
                "new_path": str(new_path),
                "new_workspace_id": new_id,
                "new_project_slug": new_slug,
            }
        )

    # orphans
    orphans_dir = ARCHIVE_ROOT / "orphans"
    if orphans_dir.exists():
        for child in orphans_dir.iterdir():
            if child.is_dir():
                copy_tree(child, new_projects / child.name)

    if (new_global / "state.vscdb").exists():
        db_stats = remap_global_db(
            new_global / "state.vscdb",
            old_home,
            str(new_home),
            repo_mappings,
        )
    else:
        db_stats = {}

    # ide_state
    ide_arch = ARCHIVE_ROOT / "global" / "ide_state.json"
    if ide_arch.exists():
        text = ide_arch.read_text(encoding="utf-8").replace(old_home, str(new_home))
        ide_dst = new_home / ".cursor" / "ide_state.json"
        ide_dst.parent.mkdir(parents=True, exist_ok=True)
        ide_dst.write_text(text, encoding="utf-8")

    report_path = new_home / ".cursor" / "cursor-chat-restore-report.json"
    report_path.write_text(
        json.dumps(
            {"db_stats": db_stats, "repos": restore_report},
            ensure_ascii=False,
            indent=2,
        )
        + "\n",
        encoding="utf-8",
    )

    print("Restore complete.")
    print("Backup of previous DB:", backup_root)
    print("Report:", report_path)
    print("DB stats:", json.dumps(db_stats, indent=2))
    print("\nRestart Cursor, then open repos under:", new_repos_root)


if __name__ == "__main__":
    main()