#!/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()