Recuperando histórico do Cursor depois de renomear projetos.

#!/usr/bin/env python3
"""Remapeia chats locais do Cursor após renomear ou mover uma pasta.

Uso (com o Cursor totalmente fechado):
  python3 cursor-remap-chats.py --old /path/antigo --new /path/novo
"""
from __future__ import annotations

import argparse
import json
import shutil
import sqlite3
import subprocess
import sys
from datetime import datetime
from pathlib import Path
from urllib.parse import quote, unquote, urlparse


def die(message: str, code: int = 1) -> None:
    print(f"ERROR: {message}", file=sys.stderr)
    raise SystemExit(code)


def cursor_running() -> bool:
    commands = (
        ["pgrep", "-x", "Cursor"],
        ["pgrep", "-f", "Cursor.app/Contents/MacOS/Cursor"],
    )
    for command in commands:
        try:
            if subprocess.check_output(command, text=True).strip():
                return True
        except (subprocess.CalledProcessError, FileNotFoundError):
            pass
    return False


def file_uri(path: str) -> str:
    return "file://" + quote(path, safe="/")


def uri_to_path(value: str) -> str | None:
    parsed = urlparse(value)
    if parsed.scheme != "file":
        return None
    return unquote(parsed.path)


def find_workspace_id(
    workspace_storage: Path,
    folder_path: str,
) -> str | None:
    target = str(Path(folder_path).expanduser().resolve())

    for workspace_json in workspace_storage.glob(
        "*/workspace.json"
    ):
        try:
            data = json.loads(
                workspace_json.read_text(encoding="utf-8")
            )
        except (OSError, json.JSONDecodeError):
            continue

        value = data.get("folder") or data.get("workspace")
        if not isinstance(value, str):
            continue

        decoded = uri_to_path(value)
        if (
            decoded
            and str(Path(decoded).expanduser().resolve())
            == target
        ):
            return workspace_json.parent.name

    return None


def rewrite_composer_value(
    raw: str,
    old_id: str,
    new_id: str,
    old_path: str,
    new_path: str,
) -> str:
    old_uri = file_uri(old_path)
    new_uri = file_uri(new_path)

    try:
        data = json.loads(raw)
    except json.JSONDecodeError:
        return (
            raw.replace(old_id, new_id)
            .replace(old_uri, new_uri)
            .replace(old_path, new_path)
        )

    workspace = data.get("workspaceIdentifier")
    if isinstance(workspace, dict):
        workspace["id"] = new_id
        uri = workspace.get("uri")
        if isinstance(uri, dict):
            uri["fsPath"] = new_path
            uri["path"] = new_path
            uri["external"] = new_uri

    repos = data.get("trackedGitRepos")
    if isinstance(repos, list):
        for repo in repos:
            if (
                isinstance(repo, dict)
                and isinstance(repo.get("repoPath"), str)
            ):
                repo["repoPath"] = repo[
                    "repoPath"
                ].replace(old_path, new_path)

    text = json.dumps(
        data,
        ensure_ascii=False,
        separators=(",", ":"),
    )

    return (
        text.replace(old_id, new_id)
        .replace(old_uri, new_uri)
        .replace(old_path, new_path)
    )


def table_exists(
    connection: sqlite3.Connection,
    table: str,
) -> bool:
    row = connection.execute(
        """
        SELECT 1
        FROM sqlite_master
        WHERE type='table' AND name=?
        """,
        (table,),
    ).fetchone()

    return row is not None


def backup_sqlite_files(
    source_dir: Path,
    destination_dir: Path,
) -> None:
    destination_dir.mkdir(parents=True, exist_ok=True)

    names = (
        "state.vscdb",
        "state.vscdb-wal",
        "state.vscdb-shm",
        "state.vscdb.backup",
    )

    for name in names:
        source = source_dir / name
        if source.exists():
            shutil.copy2(
                source,
                destination_dir / name,
            )


def sqlite_backup(
    source: Path,
    destination: Path,
) -> None:
    """Cria uma cópia consistente, incluindo o WAL."""

    with sqlite3.connect(source) as source_connection:
        with sqlite3.connect(
            destination
        ) as destination_connection:
            source_connection.backup(
                destination_connection
            )


def remap_normalized_headers(
    connection: sqlite3.Connection,
    old_id: str,
    new_id: str,
    old_path: str,
    new_path: str,
    backup_dir: Path,
) -> int:
    if not table_exists(connection, "composerHeaders"):
        return 0

    columns = {
        row[1]
        for row in connection.execute(
            "PRAGMA table_info(composerHeaders)"
        ).fetchall()
    }

    required = {
        "composerId",
        "workspaceId",
        "value",
    }

    if not required.issubset(columns):
        print(
            "  aviso: tabela composerHeaders existe, "
            "mas o schema não é reconhecido"
        )
        return 0

    rows = connection.execute(
        """
        SELECT composerId, value
        FROM composerHeaders
        WHERE workspaceId=?
        """,
        (old_id,),
    ).fetchall()

    backup_file = (
        backup_dir
        / "composerHeaders_old_workspace.json"
    )

    backup_file.write_text(
        json.dumps(
            [
                {
                    "composerId": composer_id,
                    "value": value,
                }
                for composer_id, value in rows
            ],
            ensure_ascii=False,
        ),
        encoding="utf-8",
    )

    for composer_id, value in rows:
        connection.execute(
            """
            UPDATE composerHeaders
            SET workspaceId=?, value=?
            WHERE composerId=?
            """,
            (
                new_id,
                rewrite_composer_value(
                    value,
                    old_id,
                    new_id,
                    old_path,
                    new_path,
                ),
                composer_id,
            ),
        )

    return len(rows)


def remap_itemtable_headers(
    connection: sqlite3.Connection,
    old_id: str,
    new_id: str,
    old_path: str,
    new_path: str,
) -> int:
    if not table_exists(connection, "ItemTable"):
        return 0

    row = connection.execute(
        """
        SELECT value
        FROM ItemTable
        WHERE key='composer.composerHeaders'
        """
    ).fetchone()

    if not row:
        return 0

    try:
        headers = json.loads(row[0])
    except (TypeError, json.JSONDecodeError):
        return 0

    changed = 0
    new_uri = file_uri(new_path)

    for composer in headers.get("allComposers", []):
        if not isinstance(composer, dict):
            continue

        workspace = (
            composer.get("workspaceIdentifier") or {}
        )
        uri = workspace.get("uri") or {}
        fs_path = (
            uri.get("fsPath")
            or uri.get("path")
            or ""
        )

        belongs_to_old_workspace = (
            workspace.get("id") == old_id
            or old_path in fs_path
        )

        if not belongs_to_old_workspace:
            continue

        composer["workspaceIdentifier"] = {
            "id": new_id,
            "uri": {
                "$mid": 1,
                "fsPath": new_path,
                "external": new_uri,
                "path": new_path,
                "scheme": "file",
            },
        }

        repos = composer.get("trackedGitRepos")
        if isinstance(repos, list):
            for repo in repos:
                if (
                    isinstance(repo, dict)
                    and isinstance(
                        repo.get("repoPath"),
                        str,
                    )
                ):
                    repo["repoPath"] = repo[
                        "repoPath"
                    ].replace(old_path, new_path)

        changed += 1

    if changed:
        connection.execute(
            """
            UPDATE ItemTable
            SET value=?
            WHERE key='composer.composerHeaders'
            """,
            (
                json.dumps(
                    headers,
                    ensure_ascii=False,
                    separators=(",", ":"),
                ),
            ),
        )

    return changed


def main() -> int:
    parser = argparse.ArgumentParser(
        description=(
            "Remapeia chats do Cursor após "
            "rename ou move de projeto"
        )
    )

    parser.add_argument(
        "--old",
        required=True,
        help="Path absoluto da pasta antiga",
    )
    parser.add_argument(
        "--new",
        required=True,
        help="Path absoluto da pasta nova",
    )
    parser.add_argument(
        "--also-copy-workspace-state",
        action="store_true",
        help=(
            "Copia o estado local do workspace "
            "antigo para o novo"
        ),
    )

    args = parser.parse_args()

    old_path = str(
        Path(args.old).expanduser().resolve()
    )
    new_path = str(
        Path(args.new).expanduser().resolve()
    )

    if cursor_running():
        die(
            "Cursor ainda está rodando. "
            "Feche com Cmd+Q e rode novamente."
        )

    support = (
        Path.home()
        / "Library/Application Support/Cursor/User"
    )
    workspace_storage = support / "workspaceStorage"
    global_storage = support / "globalStorage"
    global_db = global_storage / "state.vscdb"

    if not global_db.is_file():
        die(
            f"Banco global não encontrado: {global_db}"
        )

    old_id = find_workspace_id(
        workspace_storage,
        old_path,
    )
    new_id = find_workspace_id(
        workspace_storage,
        new_path,
    )

    if not old_id:
        die(
            "Não encontrei workspaceStorage "
            f"para o path antigo: {old_path}"
        )

    if not new_id:
        die(
            "Não encontrei workspaceStorage "
            f"para o path novo: {new_path}\n"
            "Abra a pasta nova uma vez no Cursor, "
            "feche o aplicativo e rode novamente."
        )

    if old_id == new_id:
        die(
            "Os paths antigo e novo apontam para "
            "o mesmo workspace id; nada a remapear."
        )

    print(
        f"old: {old_path}\n"
        f"  workspace={old_id}"
    )
    print(
        f"new: {new_path}\n"
        f"  workspace={new_id}"
    )

    stamp = datetime.now().strftime(
        "%Y%m%d_%H%M%S"
    )
    backup_dir = (
        global_storage
        / f"_backup_chat_remap_{stamp}"
    )

    backup_sqlite_files(
        global_storage,
        backup_dir / "globalStorage",
    )

    old_workspace = workspace_storage / old_id
    new_workspace = workspace_storage / new_id

    backup_sqlite_files(
        old_workspace,
        backup_dir / "old_workspace",
    )
    backup_sqlite_files(
        new_workspace,
        backup_dir / "new_workspace",
    )

    if args.also_copy_workspace_state:
        old_state = (
            old_workspace / "state.vscdb"
        )
        new_state = (
            new_workspace / "state.vscdb"
        )

        if not old_state.is_file():
            die(
                "Banco do workspace antigo "
                f"não encontrado: {old_state}"
            )

        for suffix in (
            "",
            "-wal",
            "-shm",
            ".backup",
        ):
            target = Path(
                str(new_state) + suffix
            )
            if target.exists():
                target.unlink()

        sqlite_backup(
            old_state,
            new_state,
        )

        images = old_workspace / "images"
        if images.is_dir():
            destination = (
                new_workspace / "images"
            )
            destination.mkdir(exist_ok=True)
            shutil.copytree(
                images,
                destination,
                dirs_exist_ok=True,
            )

        print(
            "→ estado do workspace copiado "
            "(abas/UI); workspace.json novo "
            "foi preservado"
        )

    with sqlite3.connect(
        global_db
    ) as connection:
        connection.execute(
            "PRAGMA busy_timeout=5000"
        )

        try:
            connection.execute(
                "PRAGMA wal_checkpoint(TRUNCATE)"
            )
        except sqlite3.Error as error:
            print(
                f"  aviso no checkpoint: {error}"
            )

        normalized = remap_normalized_headers(
            connection,
            old_id,
            new_id,
            old_path,
            new_path,
            backup_dir,
        )

        cached = remap_itemtable_headers(
            connection,
            old_id,
            new_id,
            old_path,
            new_path,
        )

        if table_exists(connection, "ItemTable"):
            old_glass = (
                "cursor/glass.tabs.v2/"
                f"{old_id}/state.json"
            )
            new_glass = (
                "cursor/glass.tabs.v2/"
                f"{new_id}/state.json"
            )

            old_glass_exists = (
                connection.execute(
                    """
                    SELECT 1
                    FROM ItemTable
                    WHERE key=?
                    """,
                    (old_glass,),
                ).fetchone()
            )

            if old_glass_exists:
                new_glass_exists = (
                    connection.execute(
                        """
                        SELECT 1
                        FROM ItemTable
                        WHERE key=?
                        """,
                        (new_glass,),
                    ).fetchone()
                )

                if new_glass_exists:
                    connection.execute(
                        """
                        DELETE FROM ItemTable
                        WHERE key=?
                        """,
                        (old_glass,),
                    )
                else:
                    connection.execute(
                        """
                        UPDATE ItemTable
                        SET key=?
                        WHERE key=?
                        """,
                        (
                            new_glass,
                            old_glass,
                        ),
                    )

                print(
                    "→ glass tabs remapeado"
                )

        inline_diffs = 0

        if table_exists(
            connection,
            "cursorDiskKV",
        ):
            rows = connection.execute(
                """
                SELECT key
                FROM cursorDiskKV
                WHERE key LIKE ?
                """,
                (
                    f"inlineDiff:{old_id}:%",
                ),
            ).fetchall()

            for (key,) in rows:
                new_key = key.replace(
                    f"inlineDiff:{old_id}:",
                    f"inlineDiff:{new_id}:",
                    1,
                )

                already_exists = (
                    connection.execute(
                        """
                        SELECT 1
                        FROM cursorDiskKV
                        WHERE key=?
                        """,
                        (new_key,),
                    ).fetchone()
                )

                if already_exists:
                    connection.execute(
                        """
                        DELETE FROM cursorDiskKV
                        WHERE key=?
                        """,
                        (key,),
                    )
                else:
                    connection.execute(
                        """
                        UPDATE cursorDiskKV
                        SET key=?
                        WHERE key=?
                        """,
                        (
                            new_key,
                            key,
                        ),
                    )

            inline_diffs = len(rows)

        print(
            "→ headers normalizados "
            f"remapeados: {normalized}"
        )
        print(
            "→ cache composer.composerHeaders "
            f"remapeado: {cached}"
        )
        print(
            "→ inlineDiff remapeados: "
            f"{inline_diffs}"
        )

        if normalized == 0 and cached == 0:
            die(
                "Nenhum chat associado ao "
                "workspace antigo foi encontrado. "
                "O caso pode ser outro: chat "
                "arquivado, sessão da Agents Window, "
                "índice corrompido ou formato ainda "
                "não suportado."
            )

        connection.commit()

        try:
            connection.execute(
                "PRAGMA wal_checkpoint(TRUNCATE)"
            )
        except sqlite3.Error as error:
            print(
                "  aviso no checkpoint final: "
                f"{error}"
            )

    cursor_backup = (
        global_storage / "state.vscdb.backup"
    )

    if cursor_backup.exists():
        shutil.copy2(
            global_db,
            cursor_backup,
        )
        print(
            "→ state.vscdb.backup existente "
            "foi sincronizado"
        )

    print()
    print(
        f"OK. backup: {backup_dir}"
    )
    print(
        "Abra o Cursor exatamente em: "
        f"{new_path}"
    )
    print(
        "Confira a lista completa de histórico, "
        "não apenas as abas abertas."
    )

    return 0


if __name__ == "__main__":
    raise SystemExit(main())