From efab050c8d0cc00b5730961a94204ba198dd7a73 Mon Sep 17 00:00:00 2001 From: Elijah Shaw-Rutschman Date: Mon, 28 Sep 2026 23:47:54 -0500 Subject: [PATCH] feat: support Claude Desktop import pointers and Claude Cloud session migration --- src/session_migrate/catalog.py | 13 + src/session_migrate/cli.py | 33 +- src/session_migrate/conversion.py | 145 ++++++++- src/session_migrate/discovery.py | 21 ++ src/session_migrate/formats/__init__.py | 2 + src/session_migrate/formats/claude.py | 200 +++++++++++- src/session_migrate/formats/claude_cloud.py | 264 +++++++++++++++ src/session_migrate/inspection.py | 25 +- src/session_migrate/model.py | 1 + tests/test_claude_cloud.py | 303 +++++++++++++++++ tests/test_claude_desktop.py | 342 ++++++++++++++++++++ tests/test_target_integration.py | 1 + 12 files changed, 1345 insertions(+), 5 deletions(-) create mode 100644 src/session_migrate/formats/claude_cloud.py create mode 100644 tests/test_claude_cloud.py create mode 100644 tests/test_claude_desktop.py diff --git a/src/session_migrate/catalog.py b/src/session_migrate/catalog.py index 2c34fe1..0f4b886 100644 --- a/src/session_migrate/catalog.py +++ b/src/session_migrate/catalog.py @@ -898,6 +898,7 @@ def add_root( AgentFormat.HERMES, AgentFormat.MASTRACODE, AgentFormat.DEVIN, + AgentFormat.CLAUDE_CLOUD, }: raise SessionMigrateError("catalog root format is unsupported") normalized = str(_absolute(path)) @@ -957,6 +958,7 @@ def refresh( hermes_roots: Sequence[Path] = (), mastracode_roots: Sequence[Path] = (), devin_roots: Sequence[Path] = (), + claude_cloud_roots: Sequence[Path] = (), discover_under: Sequence[Path] = (), include_auto: bool = True, validate: bool = False, @@ -1003,6 +1005,8 @@ def refresh( self.add_root(AgentFormat.MASTRACODE, path) for path in devin_roots: self.add_root(AgentFormat.DEVIN, path) + for path in claude_cloud_roots: + self.add_root(AgentFormat.CLAUDE_CLOUD, path) for agent_format, path, source in discover_roots(discover_under): self.add_root(agent_format, path, source=source) @@ -2005,6 +2009,14 @@ def _candidate_files(agent_format: AgentFormat, root: Path) -> Iterable[Path]: path for path in root.glob("*/events") if path.is_dir() and not path.is_symlink() ) return + if agent_format == AgentFormat.CLAUDE_CLOUD: + if root.is_file() and not root.is_symlink(): + yield root + return + yield from sorted( + path for path in root.glob("*.json") if path.is_file() and not path.is_symlink() + ) + return if agent_format == AgentFormat.CLAUDE: directories = [root / "projects"] elif agent_format == AgentFormat.CODEX: @@ -2049,6 +2061,7 @@ def _scan_file(path: Path, agent_format: AgentFormat, root: Path) -> _Scan: AgentFormat.KIMI, AgentFormat.GROK, AgentFormat.OPENHANDS, + AgentFormat.CLAUDE_CLOUD, }: return _scan_new_portable_file(path, agent_format, root) identity_labels = _native_key_labels(path, agent_format, root) diff --git a/src/session_migrate/cli.py b/src/session_migrate/cli.py index b2cc420..33b9a41 100644 --- a/src/session_migrate/cli.py +++ b/src/session_migrate/cli.py @@ -20,6 +20,7 @@ default_target_home, ensure_target_paths_available, install_antigravity_artifact, + install_claude_artifact, install_copilot_artifact, install_cursor_artifact, install_devin_artifact, @@ -42,6 +43,7 @@ shared_database_manifest_path, target_import_paths, write_artifact, + write_claude_artifact, ) from session_migrate.discovery import locate_session, normalized_source_id from session_migrate.errors import SessionMigrateError @@ -83,7 +85,11 @@ def build_parser() -> argparse.ArgumentParser: help="source transcript, native session directory, or supported export bundle", ) inspect_parser.add_argument( - "--format", choices=tuple(AgentFormat), help="override source detection" + "--from", + "--format", + dest="format", + choices=tuple(AgentFormat), + help="override source detection", ) inspect_parser.add_argument( "--json", action="store_true", help="print the structural summary as JSON" @@ -674,6 +680,20 @@ def main(argv: Sequence[str] | None = None) -> int: target_home=home, dry_run=dry_run, ) + elif target_format == TargetFormat.CLAUDE and args.command != "convert": + install_claude_artifact( + artifact, + target_home=home, + desktop_dir=getattr(args, "claude_desktop_dir", None), + dry_run=dry_run, + ) + elif args.command == "convert" and target_format == TargetFormat.CLAUDE: + write_claude_artifact( + artifact, + output_path=output_path, + manifest_path=manifest_path, + desktop_dir=getattr(args, "claude_desktop_dir", None), + ) elif not dry_run: write_artifact( artifact, @@ -731,8 +751,17 @@ def _add_conversion_arguments( ) -> None: if include_source_format: parser.add_argument( - "--format", choices=tuple(AgentFormat), help="override source detection" + "--from", + "--format", + dest="format", + choices=tuple(AgentFormat), + help="override source detection", ) + parser.add_argument( + "--claude-desktop-dir", + type=_expanded_path, + help="Claude Desktop configuration or claude-code-sessions directory", + ) parser.add_argument("--session-id", help="target UUID (generated by default)") parser.add_argument( "--cwd", type=_expanded_path, help="absolute working directory stored in the target" diff --git a/src/session_migrate/conversion.py b/src/session_migrate/conversion.py index 46d9497..6fd9759 100644 --- a/src/session_migrate/conversion.py +++ b/src/session_migrate/conversion.py @@ -22,6 +22,7 @@ from session_migrate.formats import ( antigravity, claude, + claude_cloud, codex, copilot, cursor, @@ -126,8 +127,15 @@ def manifest(self, *, output_path: Path | str) -> dict[str, Any]: } -def load_session(path: Path, source_format: AgentFormat | None = None) -> Session: +def load_session( + path: Path, + source_format: AgentFormat | None = None, + *, + session_id: str | None = None, +) -> Session: source_format = source_format or detect_path_format(path) + if source_format == AgentFormat.CLAUDE_CLOUD: + return claude_cloud.parse_session(path, session_id=session_id) if source_format == AgentFormat.ANTIGRAVITY: # The adapter makes a transactionally consistent SQLite backup that # includes live WAL state; a plain file stat/hash check would not. @@ -798,6 +806,141 @@ def write_artifact(artifact: ConversionArtifact, *, output_path: Path, manifest_ os.close(manifest_guard) +def install_claude_artifact( + artifact: ConversionArtifact, + *, + target_home: Path, + desktop_dir: Path | None = None, + dry_run: bool = False, + environ: Mapping[str, str] | None = None, +) -> tuple[Path, Path, Path | None]: + """Install Claude's native JSONL transcript and companion desktop pointer if present.""" + + if artifact.target_format != TargetFormat.CLAUDE: + raise SessionMigrateError("Claude installation requires a Claude artifact") + output_path, manifest_path = target_import_paths(artifact, target_home) + env = os.environ if environ is None else environ + sessions_dir = claude.resolve_desktop_sessions_dir( + desktop_dir=desktop_dir, + environ=env, + target_home=target_home, + ) + pointer_path: Path | None = None + if sessions_dir is not None: + pointer_path = claude.desktop_pointer_path(sessions_dir, artifact.session_id, artifact.cwd) + + check_paths: list[Path] = [output_path, manifest_path] + if pointer_path is not None: + check_paths.append(pointer_path) + ensure_target_paths_available(*check_paths) + + if dry_run: + return output_path, manifest_path, pointer_path + + manifest_bytes = ( + json.dumps(artifact.manifest(output_path=output_path), indent=2, sort_keys=True) + "\n" + ).encode() + files_to_write: list[tuple[Path, bytes]] = [ + (output_path, artifact.native_bytes), + (manifest_path, manifest_bytes), + ] + if pointer_path is not None: + pointer_bytes = claude.desktop_pointer_bytes( + session_id=artifact.session_id, + cwd=artifact.cwd, + timestamp=artifact.timestamp, + model=artifact.source.model, + title=artifact.source.title, + events=artifact.source.events, + ) + files_to_write.append((pointer_path, pointer_bytes)) + + identities: list[tuple[Path, tuple[int, int]]] = [] + guards: list[int] = [] + try: + for path, data in files_to_write: + _mkdir_private_tree(path.parent) + identity = write_private_atomic(path, data) + identities.append((path, identity)) + guards.append(_open_identity_guard(path, identity)) + if not all(_path_matches_identity(path, identity) for path, identity in identities): + raise JsonlError("Claude artifact changed during installation") + except BaseException: + for path, identity in reversed(identities): + _unlink_if_identity_matches(path, identity) + raise + finally: + for descriptor in guards: + os.close(descriptor) + return output_path, manifest_path, pointer_path + + +def write_claude_artifact( + artifact: ConversionArtifact, + *, + output_path: Path, + manifest_path: Path, + desktop_dir: Path | None = None, + environ: Mapping[str, str] | None = None, +) -> tuple[Path, Path, Path | None]: + """Write Claude native artifact and optional desktop pointer when desktop_dir or env is set.""" + + output_path = _absolute_no_follow(output_path) + manifest_path = _absolute_no_follow(manifest_path) + env = os.environ if environ is None else environ + sessions_dir = None + if desktop_dir or env.get("CLAUDE_DESKTOP_CONFIG_DIR"): + sessions_dir = claude.resolve_desktop_sessions_dir( + desktop_dir=desktop_dir, + environ=env, + ) + pointer_path: Path | None = None + if sessions_dir is not None: + pointer_path = claude.desktop_pointer_path(sessions_dir, artifact.session_id, artifact.cwd) + + check_paths: list[Path] = [output_path, manifest_path] + if pointer_path is not None: + check_paths.append(pointer_path) + ensure_target_paths_available(*check_paths) + + manifest_bytes = ( + json.dumps(artifact.manifest(output_path=output_path), indent=2, sort_keys=True) + "\n" + ).encode() + files_to_write: list[tuple[Path, bytes]] = [ + (output_path, artifact.native_bytes), + (manifest_path, manifest_bytes), + ] + if pointer_path is not None: + pointer_bytes = claude.desktop_pointer_bytes( + session_id=artifact.session_id, + cwd=artifact.cwd, + timestamp=artifact.timestamp, + model=artifact.source.model, + title=artifact.source.title, + events=artifact.source.events, + ) + files_to_write.append((pointer_path, pointer_bytes)) + + identities: list[tuple[Path, tuple[int, int]]] = [] + guards: list[int] = [] + try: + for path, data in files_to_write: + _mkdir_private_tree(path.parent) + identity = write_private_atomic(path, data) + identities.append((path, identity)) + guards.append(_open_identity_guard(path, identity)) + if not all(_path_matches_identity(path, identity) for path, identity in identities): + raise JsonlError("Claude artifact changed during installation") + except BaseException: + for path, identity in reversed(identities): + _unlink_if_identity_matches(path, identity) + raise + finally: + for descriptor in guards: + os.close(descriptor) + return output_path, manifest_path, pointer_path + + def install_copilot_artifact( artifact: ConversionArtifact, *, diff --git a/src/session_migrate/discovery.py b/src/session_migrate/discovery.py index 0545fed..f89803a 100644 --- a/src/session_migrate/discovery.py +++ b/src/session_migrate/discovery.py @@ -12,6 +12,7 @@ from session_migrate.errors import SessionMigrateError from session_migrate.formats import ( claude, + claude_cloud, cursor, devin, grok, @@ -95,6 +96,9 @@ def locate_session( _reject_central_database_cwd(cwd, "Devin") database = devin.database_path(home) matches = _central_database_match(database, normalized_id, devin.list_sessions) + elif source_format == AgentFormat.CLAUDE_CLOUD: + _reject_central_database_cwd(cwd, "Claude Cloud") + matches = _claude_cloud_matches(home, normalized_id) else: raise SessionMigrateError( "OpenCode and Kilo sessions are exported through their official CLIs, " @@ -223,6 +227,23 @@ def _grok_matches(home: Path, session_id: str, cwd: Path | None) -> list[Path]: return [path.parent for path in sessions.glob(f"*/{session_id}/summary.json")] +def _claude_cloud_matches(home: Path, session_id: str) -> list[Path]: + if home.is_file(): + return [home] if claude_cloud.has_session(home, session_id) else [] + matches: list[Path] = [] + conversations_file = home / "conversations.json" + if conversations_file.is_file() and claude_cloud.has_session(conversations_file, session_id): + matches.append(conversations_file) + for json_file in home.glob("*.json"): + if ( + json_file != conversations_file + and json_file.is_file() + and claude_cloud.has_session(json_file, session_id) + ): + matches.append(json_file) + return matches + + def normalized_session_id(value: str) -> str: try: return str(uuid.UUID(value)) diff --git a/src/session_migrate/formats/__init__.py b/src/session_migrate/formats/__init__.py index 1600409..3d4b54b 100644 --- a/src/session_migrate/formats/__init__.py +++ b/src/session_migrate/formats/__init__.py @@ -3,6 +3,7 @@ from session_migrate.formats import ( antigravity, claude, + claude_cloud, codex, copilot, cursor, @@ -24,6 +25,7 @@ __all__ = [ "antigravity", "claude", + "claude_cloud", "codex", "copilot", "cursor", diff --git a/src/session_migrate/formats/claude.py b/src/session_migrate/formats/claude.py index 5a58e58..60e982c 100644 --- a/src/session_migrate/formats/claude.py +++ b/src/session_migrate/formats/claude.py @@ -2,10 +2,13 @@ from __future__ import annotations +import json +import os import re +import sys import uuid from collections import Counter, deque -from collections.abc import Iterable +from collections.abc import Iterable, Mapping from datetime import UTC, datetime from pathlib import Path from typing import Any @@ -678,3 +681,198 @@ def _omission_key(event: Event) -> str: def _utc_now() -> str: return datetime.now(UTC).isoformat().replace("+00:00", "Z") + + +DEFAULT_DESKTOP_ORG_ID = "00000000-0000-4000-8000-000000000001" +DEFAULT_DESKTOP_PROJECT_ID = "00000000-0000-4000-8000-000000000001" +DEFAULT_DESKTOP_MODEL = "claude-3-7-sonnet-20250219" + + +def resolve_desktop_sessions_dir( + desktop_dir: Path | None = None, + environ: Mapping[str, str] | None = None, + target_home: Path | None = None, +) -> Path | None: + """Resolve the active Claude Desktop claude-code-sessions directory. + + Checks: + 1. Explicit desktop_dir argument. + 2. CLAUDE_DESKTOP_CONFIG_DIR in environ. + 3. claude-code-sessions subdirectory inside target_home (if provided). + 4. Default OS paths (if present on disk): + - macOS: ~/Library/Application Support/Claude/claude-code-sessions + - Linux: $XDG_CONFIG_HOME/Claude/claude-code-sessions or + ~/.config/Claude/claude-code-sessions + - Windows: %APPDATA%/Claude/claude-code-sessions + """ + env = os.environ if environ is None else environ + if desktop_dir is not None: + p = Path(desktop_dir).expanduser().resolve() + return p if p.name == "claude-code-sessions" else p / "claude-code-sessions" + + env_dir = env.get("CLAUDE_DESKTOP_CONFIG_DIR") + if env_dir: + p = Path(env_dir).expanduser().resolve() + return p if p.name == "claude-code-sessions" else p / "claude-code-sessions" + + if target_home is not None: + candidate = target_home.expanduser().resolve() / "claude-code-sessions" + if candidate.is_dir(): + return candidate + + if sys.platform == "darwin": + default_path = ( + Path.home() / "Library" / "Application Support" / "Claude" / "claude-code-sessions" + ) + elif sys.platform == "win32": + app_data = env.get("APPDATA") + default_path = ( + (Path(app_data).expanduser() / "Claude" / "claude-code-sessions") + if app_data + else Path.home() / "AppData" / "Roaming" / "Claude" / "claude-code-sessions" + ) + else: + xdg_config = env.get("XDG_CONFIG_HOME") + default_path = ( + (Path(xdg_config).expanduser() / "Claude" / "claude-code-sessions") + if xdg_config + else Path.home() / ".config" / "Claude" / "claude-code-sessions" + ) + + if default_path.is_dir(): + return default_path.resolve() + return None + + +def find_desktop_project_dir(sessions_dir: Path, session_cwd: Path | None = None) -> Path: + """Find the matching or primary / directory under claude-code-sessions.""" + pairs: list[tuple[Path, Path]] = [] + matching_pair: tuple[Path, Path] | None = None + norm_cwd = str(session_cwd.resolve()) if session_cwd else None + + if sessions_dir.is_dir(): + for org in sorted(sessions_dir.iterdir()): + if org.is_dir() and not org.name.startswith("."): + for proj in sorted(org.iterdir()): + if proj.is_dir() and not proj.name.startswith("."): + pairs.append((org, proj)) + if matching_pair is None and norm_cwd: + for pointer_file in proj.glob("local_*.json"): + try: + data = json.loads(pointer_file.read_text(encoding="utf-8")) + p_cwd = data.get("cwd") or data.get("originCwd") + if p_cwd and ( + p_cwd == norm_cwd or str(Path(p_cwd).resolve()) == norm_cwd + ): + matching_pair = (org, proj) + break + except Exception: + continue + + if matching_pair is not None: + return matching_pair[1] + + if pairs: + def pair_key(pair: tuple[Path, Path]) -> tuple[float, str, str]: + org, proj = pair + latest_mtime = 0.0 + for f in proj.glob("local_*.json"): + try: + m = f.stat().st_mtime + if m > latest_mtime: + latest_mtime = m + except OSError: + pass + return (latest_mtime, org.name, proj.name) + + sorted_pairs = sorted(pairs, key=pair_key, reverse=True) + return sorted_pairs[0][1] + + orgs = ( + [p for p in sorted(sessions_dir.iterdir()) if p.is_dir() and not p.name.startswith(".")] + if sessions_dir.is_dir() + else [] + ) + if orgs: + return orgs[0] / DEFAULT_DESKTOP_PROJECT_ID + return sessions_dir / DEFAULT_DESKTOP_ORG_ID / DEFAULT_DESKTOP_PROJECT_ID + + +def desktop_pointer_path( + sessions_dir: Path, session_id: str, session_cwd: Path | None = None +) -> Path: + """Return the absolute path for local_.json.""" + return find_desktop_project_dir(sessions_dir, session_cwd) / f"local_{session_id}.json" + + +def parse_timestamp_ms(timestamp: str | None) -> int: + """Parse an RFC 3339 / ISO timestamp string to integer milliseconds since epoch.""" + if timestamp: + try: + cleaned = timestamp.replace("Z", "+00:00") + dt = datetime.fromisoformat(cleaned) + return int(dt.timestamp() * 1000) + except Exception: + pass + return int(datetime.now(UTC).timestamp() * 1000) + + +def desktop_pointer_data( + session_id: str, + cwd: Path, + timestamp: str | None = None, + model: str | None = None, + title: str | None = None, + events: Iterable[Event] = (), +) -> dict[str, Any]: + """Generate the JSON payload for companion Claude Desktop pointer file + local_.json. + """ + created_at_ms = parse_timestamp_ms(timestamp) + last_activity_ms = created_at_ms + + for event in events: + if event.timestamp: + event_ms = parse_timestamp_ms(event.timestamp) + if event_ms > last_activity_ms: + last_activity_ms = event_ms + + cwd_str = str(cwd.resolve()) + + return { + "sessionId": f"local_{session_id}", + "cliSessionId": session_id, + "cwd": cwd_str, + "originCwd": cwd_str, + "lastFocusedAt": last_activity_ms, + "createdAt": created_at_ms, + "lastActivityAt": last_activity_ms, + "model": model or DEFAULT_DESKTOP_MODEL, + "isArchived": False, + "title": title or "Migrated Session", + "titleSource": "custom", + "permissionMode": "default", + "importedFrom": "local-1p-code", + "alwaysAllowedReasons": [], + "sessionPermissionUpdates": [], + } + + +def desktop_pointer_bytes( + session_id: str, + cwd: Path, + timestamp: str | None = None, + model: str | None = None, + title: str | None = None, + events: Iterable[Event] = (), +) -> bytes: + """Return formatted JSON bytes for local_.json.""" + payload = desktop_pointer_data( + session_id=session_id, + cwd=cwd, + timestamp=timestamp, + model=model, + title=title, + events=events, + ) + return (json.dumps(payload, indent=2, sort_keys=True) + "\n").encode() diff --git a/src/session_migrate/formats/claude_cloud.py b/src/session_migrate/formats/claude_cloud.py new file mode 100644 index 0000000..49cbb62 --- /dev/null +++ b/src/session_migrate/formats/claude_cloud.py @@ -0,0 +1,264 @@ +"""Claude Cloud (Claude.ai) export reader and session adapter.""" + +from __future__ import annotations + +import json +import uuid +from pathlib import Path +from typing import Any + +from session_migrate.errors import SessionMigrateError +from session_migrate.formats.common import content_text, string, valid_rfc3339 +from session_migrate.jsonl import DEFAULT_MAX_TOTAL_BYTES, file_sha256 +from session_migrate.model import AgentFormat, Event, EventKind, Provenance, Role, Session + +PINNED_CLAUDE_CLOUD_VERSION = "claude-cloud" + + +def load_conversations(path: Path) -> list[dict[str, Any]]: + """Load conversation dictionaries from a Claude.ai export JSON file.""" + if not path.is_file() or path.is_symlink(): + raise SessionMigrateError(f"Claude Cloud source path must be a regular file: {path}") + try: + size = path.stat().st_size + except OSError as exc: + raise SessionMigrateError(f"cannot stat Claude Cloud source file: {path}") from exc + if size > DEFAULT_MAX_TOTAL_BYTES: + raise SessionMigrateError( + f"Claude Cloud export file exceeds size limit ({size} > {DEFAULT_MAX_TOTAL_BYTES})" + ) + try: + data = json.loads(path.read_text(encoding="utf-8")) + except (OSError, UnicodeDecodeError, json.JSONDecodeError, ValueError) as exc: + raise SessionMigrateError(f"cannot parse Claude Cloud export file {path}: {exc}") from exc + + if isinstance(data, list): + return [c for c in data if isinstance(c, dict)] + if isinstance(data, dict): + if "conversations" in data and isinstance(data["conversations"], list): + return [c for c in data["conversations"] if isinstance(c, dict)] + if "chat_messages" in data or "uuid" in data: + return [data] + raise SessionMigrateError(f"unrecognized Claude Cloud export structure in {path}") + + +def is_claude_cloud_document(data: Any) -> bool: + """Determine if a parsed JSON structure matches Claude Cloud exports.""" + if isinstance(data, list): + if not data: + return False + first = data[0] + return bool( + isinstance(first, dict) + and "chat_messages" in first + and ("uuid" in first or "name" in first or "created_at" in first) + ) + if isinstance(data, dict): + if "chat_messages" in data and ( + "uuid" in data or "name" in data or "created_at" in data + ): + return True + convs = data.get("conversations") + if isinstance(convs, list) and convs: + first = convs[0] + if isinstance(first, dict) and "chat_messages" in first: + return True + return False + + +def is_claude_cloud_path(path: Path, size: int) -> bool: + """Check whether a file on disk is a Claude Cloud JSON export.""" + if size > DEFAULT_MAX_TOTAL_BYTES or size == 0: + return False + try: + data = json.loads(path.read_bytes()) + except (OSError, UnicodeDecodeError, json.JSONDecodeError, ValueError): + return False + return is_claude_cloud_document(data) + + +def has_session(path: Path, session_id: str) -> bool: + """Return whether the file contains a conversation with the specified session ID.""" + try: + conversations = load_conversations(path) + except Exception: + return False + norm_id = session_id.lower() + for conv in conversations: + conv_id = string(conv.get("uuid")) or string(conv.get("id")) + if conv_id and conv_id.lower() == norm_id: + return True + return False + + +def list_sessions(path: Path) -> tuple[dict[str, Any], ...]: + """Return metadata summaries for all conversations in a Claude Cloud export.""" + conversations = load_conversations(path) + summaries: list[dict[str, Any]] = [] + for conv in conversations: + conv_id = string(conv.get("uuid")) or string(conv.get("id")) or "" + title = string(conv.get("name")) or string(conv.get("title")) or "Untitled" + created_at = valid_rfc3339(conv.get("created_at")) + messages = conv.get("chat_messages") or conv.get("messages") or [] + summaries.append( + { + "session_id": conv_id, + "title": title, + "created_at": created_at, + "records": len(messages) if isinstance(messages, list) else 0, + } + ) + return tuple(summaries) + + +def parse_session(path: Path, session_id: str | None = None) -> Session: + """Parse one Claude Cloud conversation from an export JSON file into a Session.""" + conversations = load_conversations(path) + if not conversations: + raise SessionMigrateError(f"Claude Cloud export contains no conversations: {path}") + + selected: dict[str, Any] | None = None + if session_id: + norm_id = session_id.lower() + for conv in conversations: + conv_id = string(conv.get("uuid")) or string(conv.get("id")) + if conv_id and conv_id.lower() == norm_id: + selected = conv + break + if selected is None: + raise SessionMigrateError( + f"no Claude Cloud conversation found for UUID {session_id} in {path}" + ) + else: + selected = conversations[0] + + conv_uuid = string(selected.get("uuid")) or string(selected.get("id")) + if not conv_uuid: + conv_uuid = str(uuid.uuid4()) + title = string(selected.get("name")) or string(selected.get("title")) + created_at = string(selected.get("created_at")) + updated_at = string(selected.get("updated_at")) + started_at = valid_rfc3339(created_at) or valid_rfc3339(updated_at) + model = string(selected.get("model")) + + chat_messages: list[Any] = ( + selected.get("chat_messages") + if isinstance(selected.get("chat_messages"), list) + else selected.get("messages") + if isinstance(selected.get("messages"), list) + else [] + ) + + events: list[Event] = [] + for idx, msg in enumerate(chat_messages): + if not isinstance(msg, dict): + continue + msg_id = string(msg.get("uuid")) or string(msg.get("id")) or str(uuid.uuid4()) + sender = string(msg.get("sender")) or string(msg.get("role")) or "unknown" + sender_lower = sender.lower() + if sender_lower in {"human", "user"}: + role = Role.USER + elif sender_lower in {"assistant", "bot"}: + role = Role.ASSISTANT + elif sender_lower == "system": + role = Role.SYSTEM + else: + role = ( + Role.USER + if "human" in sender_lower or "user" in sender_lower + else Role.ASSISTANT + ) + + if role == Role.ASSISTANT and model is None and msg.get("model"): + model = string(msg.get("model")) + + msg_ts = ( + valid_rfc3339(msg.get("created_at")) + or valid_rfc3339(msg.get("updated_at")) + or started_at + ) + text = string(msg.get("text")) or content_text(msg.get("content")) + attachments = msg.get("attachments") if isinstance(msg.get("attachments"), list) else [] + files = msg.get("files") if isinstance(msg.get("files"), list) else [] + + if not text: + extracted = [ + string(a.get("extracted_content")) + for a in attachments + if isinstance(a, dict) and string(a.get("extracted_content")) + ] + if extracted: + text = "\n\n".join(extracted) + else: + att_names = [ + string(a.get("file_name")) + for a in attachments + if isinstance(a, dict) and string(a.get("file_name")) + ] + if att_names: + text = f"[Attached: {', '.join(att_names)}]" + if not text and files: + file_names = [ + string(f.get("file_name")) + for f in files + if isinstance(f, dict) and string(f.get("file_name")) + ] + if file_names: + text = f"[Attached: {', '.join(file_names)}]" + + provenance = Provenance( + record_index=idx, + record_type=f"chat_messages:{sender}", + source_id=msg_id, + ) + payload: dict[str, Any] = {} + if attachments: + payload["attachments"] = attachments + if files: + payload["files"] = files + + events.append( + Event( + kind=EventKind.MESSAGE, + role=role, + text=text or "", + timestamp=msg_ts, + provenance=provenance, + payload=payload, + ) + ) + + for att_idx, att in enumerate(attachments): + if isinstance(att, dict): + file_type = string(att.get("file_type")) or "" + image_url = string(att.get("image_url")) or string(att.get("url")) + if image_url and file_type.startswith("image/"): + events.append( + Event( + kind=EventKind.CONTEXT, + role=role, + timestamp=msg_ts, + payload={"block_type": "image", "image_url": image_url}, + provenance=Provenance( + record_index=idx, + record_type="attachment", + source_id=msg_id, + block_index=att_idx, + ), + ) + ) + + return Session( + source_format=AgentFormat.CLAUDE_CLOUD, + source_path=path.resolve(), + source_sha256=file_sha256(path), + session_id=conv_uuid, + cwd=None, + started_at=started_at, + cli_version=None, + model=model, + title=title, + events=tuple(events), + raw_record_count=len(chat_messages), + model_provider="anthropic", + ) diff --git a/src/session_migrate/inspection.py b/src/session_migrate/inspection.py index 05040d2..ae1eb4d 100644 --- a/src/session_migrate/inspection.py +++ b/src/session_migrate/inspection.py @@ -12,6 +12,7 @@ from session_migrate.errors import FormatDetectionError, JsonlError, SessionMigrateError from session_migrate.formats import ( antigravity, + claude_cloud, cursor, devin, grok, @@ -149,6 +150,8 @@ def inspect_session(path: Path, *, source_format: AgentFormat | None = None) -> return _inspect_portable_database(mastracode.parse_session(path)) if source_format == AgentFormat.DEVIN: return _inspect_portable_database(devin.parse_session(path)) + if source_format == AgentFormat.CLAUDE_CLOUD: + return _inspect_portable_database(claude_cloud.parse_session(path)) if source_format is None and _has_sqlite_header(path): detected = _detect_sqlite_path(path) if detected == AgentFormat.ANTIGRAVITY: @@ -163,7 +166,14 @@ def inspect_session(path: Path, *, source_format: AgentFormat | None = None) -> parsed = devin.parse_session(path) return _inspect_portable_database(parsed) before = file_snapshot(path) - if source_format in {AgentFormat.OPENCODE, AgentFormat.KILO} or source_format is None: + if ( + source_format in {AgentFormat.OPENCODE, AgentFormat.KILO, AgentFormat.CLAUDE_CLOUD} + or source_format is None + ): + if claude_cloud.is_claude_cloud_path(path, before.size): + result = _inspect_portable_database(claude_cloud.parse_session(path)) + ensure_file_unchanged(path, before) + return result document = _load_json_document(path, before.size) if document is not None and source_format is None and _is_opencode_document(document): ensure_file_unchanged(path, before) @@ -352,6 +362,7 @@ def detect_format(records: list[dict[str, Any] | Any]) -> AgentFormat: muse_decisive = False qwen_decisive = False kimi_decisive = False + claude_cloud_decisive = False claude_score = 0 codex_score = 0 pi_score = 0 @@ -433,6 +444,12 @@ def detect_format(records: list[dict[str, Any] | Any]) -> AgentFormat: claude_score += 1 if "sessionId" in value or "parentUuid" in value: claude_score += 3 + if ( + record_type is None + and "chat_messages" in value + and any(k in value for k in ("uuid", "name", "created_at")) + ): + claude_cloud_decisive = True decisive = sum( ( claude_decisive, @@ -444,6 +461,7 @@ def detect_format(records: list[dict[str, Any] | Any]) -> AgentFormat: muse_decisive, qwen_decisive, kimi_decisive, + claude_cloud_decisive, ) ) if decisive > 1: @@ -466,6 +484,8 @@ def detect_format(records: list[dict[str, Any] | Any]) -> AgentFormat: return AgentFormat.QWEN if kimi_decisive: return AgentFormat.KIMI + if claude_cloud_decisive: + return AgentFormat.CLAUDE_CLOUD if pi_score and pi_score > max(claude_score, codex_score): return AgentFormat.PI if codex_score and codex_score > claude_score: @@ -504,6 +524,9 @@ def detect_path_format(path: Path) -> AgentFormat: if _has_sqlite_header(path): return _detect_sqlite_path(path) before = file_snapshot(path) + if claude_cloud.is_claude_cloud_path(path, before.size): + ensure_file_unchanged(path, before) + return AgentFormat.CLAUDE_CLOUD document = _load_json_document(path, before.size) if document is not None and _is_opencode_document(document): ensure_file_unchanged(path, before) diff --git a/src/session_migrate/model.py b/src/session_migrate/model.py index 14eead0..32ec234 100644 --- a/src/session_migrate/model.py +++ b/src/session_migrate/model.py @@ -28,6 +28,7 @@ class AgentFormat(StrEnum): HERMES = "hermes" MASTRACODE = "mastracode" DEVIN = "devin" + CLAUDE_CLOUD = "claude-cloud" class TargetFormat(StrEnum): diff --git a/tests/test_claude_cloud.py b/tests/test_claude_cloud.py new file mode 100644 index 0000000..661f06b --- /dev/null +++ b/tests/test_claude_cloud.py @@ -0,0 +1,303 @@ +import json +from pathlib import Path + +import pytest + +from session_migrate.catalog import Catalog +from session_migrate.cli import main +from session_migrate.conversion import ( + ConversionOptions, + convert_session, + load_session, +) +from session_migrate.discovery import locate_session +from session_migrate.errors import SessionMigrateError +from session_migrate.formats import claude_cloud +from session_migrate.inspection import detect_format, detect_path_format, inspect_session +from session_migrate.model import ( + AgentFormat, + EventKind, + Role, + TargetFormat, +) + +CONV_UUID_1 = "11111111-aaaa-4111-8111-111111111111" +CONV_UUID_2 = "22222222-bbbb-4222-8222-222222222222" + + +def _sample_cloud_export() -> list[dict[str, object]]: + return [ + { + "uuid": CONV_UUID_1, + "name": "First Cloud Conversation", + "created_at": "2026-08-17T10:00:00Z", + "updated_at": "2026-08-17T10:30:00Z", + "model": "claude-3-5-sonnet-20241022", + "chat_messages": [ + { + "uuid": "msg-001", + "sender": "human", + "text": "Can you explain recursion?", + "created_at": "2026-08-17T10:00:00Z", + "updated_at": "2026-08-17T10:00:00Z", + "attachments": [], + "files": [], + }, + { + "uuid": "msg-002", + "sender": "assistant", + "text": "Recursion is when a function calls itself.", + "created_at": "2026-08-17T10:01:00Z", + "updated_at": "2026-08-17T10:01:00Z", + "attachments": [], + "files": [], + }, + ], + }, + { + "uuid": CONV_UUID_2, + "name": "Second Cloud Conversation with Image and Attachment", + "created_at": "2026-08-18T14:00:00Z", + "updated_at": "2026-08-18T14:15:00Z", + "chat_messages": [ + { + "uuid": "msg-003", + "sender": "human", + "text": "", + "created_at": "2026-08-18T14:00:00Z", + "attachments": [ + { + "file_name": "data.csv", + "file_type": "text/csv", + "extracted_content": "col1,col2\nval1,val2", + }, + { + "file_name": "diagram.png", + "file_type": "image/png", + "image_url": "https://example.com/diagram.png", + }, + ], + }, + { + "uuid": "msg-004", + "sender": "assistant", + "text": "I received your CSV data and diagram.", + "model": "claude-3-opus-20240229", + "created_at": "2026-08-18T14:02:00Z", + }, + ], + }, + ] + + +def test_is_claude_cloud_document() -> None: + data = _sample_cloud_export() + assert claude_cloud.is_claude_cloud_document(data) is True + + # Single conversation dict + assert claude_cloud.is_claude_cloud_document(data[0]) is True + + # Wrapped in conversations list + assert claude_cloud.is_claude_cloud_document({"conversations": data}) is True + + # Negative cases + assert claude_cloud.is_claude_cloud_document([]) is False + assert claude_cloud.is_claude_cloud_document({}) is False + assert claude_cloud.is_claude_cloud_document({"foo": "bar"}) is False + assert claude_cloud.is_claude_cloud_document("string") is False + + +def test_is_claude_cloud_path_and_format_detection(tmp_path: Path) -> None: + export_file = tmp_path / "conversations.json" + export_file.write_text(json.dumps(_sample_cloud_export())) + + assert claude_cloud.is_claude_cloud_path(export_file, export_file.stat().st_size) is True + assert detect_path_format(export_file) == AgentFormat.CLAUDE_CLOUD + assert detect_format(_sample_cloud_export()) == AgentFormat.CLAUDE_CLOUD + + +def test_load_conversations_and_listing(tmp_path: Path) -> None: + export_file = tmp_path / "conversations.json" + export_file.write_text(json.dumps(_sample_cloud_export())) + + convs = claude_cloud.load_conversations(export_file) + assert len(convs) == 2 + + assert claude_cloud.has_session(export_file, CONV_UUID_1) is True + assert claude_cloud.has_session(export_file, CONV_UUID_2) is True + assert claude_cloud.has_session(export_file, "nonexistent-uuid") is False + + summaries = claude_cloud.list_sessions(export_file) + assert len(summaries) == 2 + assert summaries[0]["session_id"] == CONV_UUID_1 + assert summaries[0]["title"] == "First Cloud Conversation" + assert summaries[0]["records"] == 2 + assert summaries[1]["session_id"] == CONV_UUID_2 + + +def test_parse_session_default_first(tmp_path: Path) -> None: + export_file = tmp_path / "conversations.json" + export_file.write_text(json.dumps(_sample_cloud_export())) + + session = claude_cloud.parse_session(export_file) + assert session.source_format == AgentFormat.CLAUDE_CLOUD + assert session.session_id == CONV_UUID_1 + assert session.title == "First Cloud Conversation" + assert session.model == "claude-3-5-sonnet-20241022" + assert session.started_at == "2026-08-17T10:00:00Z" + assert len(session.events) == 2 + + assert session.events[0].kind == EventKind.MESSAGE + assert session.events[0].role == Role.USER + assert session.events[0].text == "Can you explain recursion?" + + assert session.events[1].kind == EventKind.MESSAGE + assert session.events[1].role == Role.ASSISTANT + assert session.events[1].text == "Recursion is when a function calls itself." + + +def test_parse_session_by_uuid_and_attachments(tmp_path: Path) -> None: + export_file = tmp_path / "conversations.json" + export_file.write_text(json.dumps(_sample_cloud_export())) + + session = claude_cloud.parse_session(export_file, session_id=CONV_UUID_2) + assert session.session_id == CONV_UUID_2 + assert session.title == "Second Cloud Conversation with Image and Attachment" + assert session.model == "claude-3-opus-20240229" + + # Message 1 had extracted text from CSV and an image attachment (emitted as CONTEXT event) + # Message 2 was assistant response + assert len(session.events) == 3 + event_user_msg = session.events[0] + assert event_user_msg.kind == EventKind.MESSAGE + assert event_user_msg.role == Role.USER + assert "col1,col2" in event_user_msg.text + + event_image = session.events[1] + assert event_image.kind == EventKind.CONTEXT + assert event_image.payload.get("block_type") == "image" + assert event_image.payload.get("image_url") == "https://example.com/diagram.png" + + event_asst_msg = session.events[2] + assert event_asst_msg.kind == EventKind.MESSAGE + assert event_asst_msg.role == Role.ASSISTANT + assert event_asst_msg.text == "I received your CSV data and diagram." + + +def test_parse_session_single_conversation_object(tmp_path: Path) -> None: + single = _sample_cloud_export()[0] + export_file = tmp_path / "single_conversation.json" + export_file.write_text(json.dumps(single)) + + session = claude_cloud.parse_session(export_file) + assert session.session_id == CONV_UUID_1 + assert len(session.events) == 2 + + +def test_parse_session_nonexistent_raises(tmp_path: Path) -> None: + export_file = tmp_path / "conversations.json" + export_file.write_text(json.dumps(_sample_cloud_export())) + + with pytest.raises(SessionMigrateError, match="no Claude Cloud conversation found"): + claude_cloud.parse_session(export_file, session_id="missing-uuid") + + +def test_load_session_and_inspection(tmp_path: Path) -> None: + export_file = tmp_path / "conversations.json" + export_file.write_text(json.dumps(_sample_cloud_export())) + + session = load_session( + export_file, + source_format=AgentFormat.CLAUDE_CLOUD, + session_id=CONV_UUID_2, + ) + assert session.session_id == CONV_UUID_2 + + inspection = inspect_session(export_file) + assert inspection.format == AgentFormat.CLAUDE_CLOUD.value + assert inspection.session_id == CONV_UUID_1 + assert inspection.records == 2 + + +def test_discovery_and_catalog(tmp_path: Path) -> None: + export_file = tmp_path / "conversations.json" + export_file.write_text(json.dumps(_sample_cloud_export())) + + located = locate_session(AgentFormat.CLAUDE_CLOUD, CONV_UUID_1, tmp_path) + assert located.resolve() == export_file.resolve() + + catalog = Catalog(tmp_path / "catalog.sqlite3") + result = catalog.refresh(claude_cloud_roots=[tmp_path], include_auto=False) + assert result.files_seen >= 1 + sessions = catalog.list_sessions(query="First Cloud Conversation") + assert len(sessions) >= 1 + assert sessions[0].session_id == CONV_UUID_1 + + +def test_convert_claude_cloud_to_codex_and_claude(tmp_path: Path) -> None: + export_file = tmp_path / "conversations.json" + export_file.write_text(json.dumps(_sample_cloud_export())) + + session = claude_cloud.parse_session(export_file, session_id=CONV_UUID_1) + + # Convert to CODEX + codex_artifact = convert_session( + session, + ConversionOptions( + target_format=TargetFormat.CODEX, + session_id="33333333-3333-4333-8333-333333333333", + cwd=tmp_path, + ), + ) + assert codex_artifact.target_format == TargetFormat.CODEX + assert codex_artifact.native_record_count > 0 + + # Convert to CLAUDE + claude_artifact = convert_session( + session, + ConversionOptions( + target_format=TargetFormat.CLAUDE, + session_id="44444444-4444-4444-8444-444444444444", + cwd=tmp_path, + ), + ) + assert claude_artifact.target_format == TargetFormat.CLAUDE + assert claude_artifact.native_record_count > 0 + + +def test_cli_inspect_and_convert(tmp_path: Path, capsys: pytest.CaptureFixture[str]) -> None: + export_file = tmp_path / "conversations.json" + export_file.write_text(json.dumps(_sample_cloud_export())) + + # CLI inspect + code = main(["inspect", str(export_file)]) + assert code == 0 + captured = capsys.readouterr() + assert "claude-cloud" in captured.out + assert CONV_UUID_1 in captured.out + + # CLI inspect with --from alias + code_from = main(["inspect", str(export_file), "--from", "claude-cloud"]) + assert code_from == 0 + + # CLI convert + out_file = tmp_path / "converted.jsonl" + convert_code = main( + [ + "convert", + str(export_file), + "--from", + "claude-cloud", + "--to", + "codex", + "--output", + str(out_file), + "--cwd", + str(tmp_path), + "--session-id", + "55555555-5555-4555-8555-555555555555", + ] + ) + assert convert_code == 0 + assert out_file.is_file() + assert (tmp_path / "converted.jsonl.session-migrate.json").is_file() diff --git a/tests/test_claude_desktop.py b/tests/test_claude_desktop.py new file mode 100644 index 0000000..16d2e8c --- /dev/null +++ b/tests/test_claude_desktop.py @@ -0,0 +1,342 @@ +import json +import os +import time +from pathlib import Path + +import pytest + +from session_migrate.cli import main +from session_migrate.conversion import ( + ConversionOptions, + convert_session, + install_claude_artifact, + write_claude_artifact, +) +from session_migrate.errors import SessionMigrateError +from session_migrate.formats import claude +from session_migrate.model import ( + AgentFormat, + Event, + EventKind, + Provenance, + Role, + Session, + TargetFormat, +) + +FIXTURES = Path(__file__).parent / "fixtures" +SOURCE_CLAUDE_FILE = FIXTURES / "claude-2.1.209" / "basic.jsonl" +TEST_SESSION_ID = "22222222-2222-4222-8222-222222222222" + + +def _sample_session(tmp_path: Path) -> Session: + return Session( + source_format=AgentFormat.CLAUDE, + source_path=tmp_path / "sample.jsonl", + source_sha256="abc123", + session_id=TEST_SESSION_ID, + cwd=tmp_path / "my_project", + started_at="2026-08-17T12:00:00Z", + cli_version="2.1.209", + model="claude-3-5-sonnet-20241022", + title="Test Desktop Session", + events=( + Event( + kind=EventKind.MESSAGE, + role=Role.USER, + text="Hello Claude Desktop", + timestamp="2026-08-17T12:00:00Z", + provenance=Provenance(record_index=0, record_type="user", source_id="u1"), + ), + Event( + kind=EventKind.MESSAGE, + role=Role.ASSISTANT, + text="Hello human!", + timestamp="2026-08-17T12:05:00Z", + provenance=Provenance(record_index=1, record_type="assistant", source_id="a1"), + ), + ), + raw_record_count=2, + ) + + +def test_resolve_desktop_sessions_dir_explicit(tmp_path: Path) -> None: + desktop_root = tmp_path / "Claude" + desktop_root.mkdir() + + # Pass root directory (without trailing claude-code-sessions) + resolved = claude.resolve_desktop_sessions_dir(desktop_dir=desktop_root) + assert resolved == (desktop_root / "claude-code-sessions").resolve() + + # Pass directory already ending in claude-code-sessions + sessions_dir = desktop_root / "claude-code-sessions" + sessions_dir.mkdir() + resolved_direct = claude.resolve_desktop_sessions_dir(desktop_dir=sessions_dir) + assert resolved_direct == sessions_dir.resolve() + + +def test_resolve_desktop_sessions_dir_environ(tmp_path: Path) -> None: + custom_dir = tmp_path / "custom_desktop" + custom_dir.mkdir() + + env = {"CLAUDE_DESKTOP_CONFIG_DIR": str(custom_dir)} + resolved = claude.resolve_desktop_sessions_dir(environ=env) + assert resolved == (custom_dir / "claude-code-sessions").resolve() + + +def test_resolve_desktop_sessions_dir_target_home(tmp_path: Path) -> None: + home = tmp_path / "target_home" + sessions = home / "claude-code-sessions" + sessions.mkdir(parents=True) + + # Isolated environ preventing OS path detection + env = {"HOME": str(tmp_path / "empty_home"), "CLAUDE_DESKTOP_CONFIG_DIR": ""} + resolved = claude.resolve_desktop_sessions_dir(environ=env, target_home=home) + assert resolved == sessions.resolve() + + +def test_resolve_desktop_sessions_dir_returns_none_when_absent(tmp_path: Path) -> None: + import sys + + empty_home = tmp_path / "empty_home" + empty_home.mkdir() + env = { + "HOME": str(empty_home), + "XDG_CONFIG_HOME": str(empty_home), + "APPDATA": str(empty_home), + } + fake_nonexistent = tmp_path / "nonexistent" + orig_platform = sys.platform + try: + sys.platform = "linux" + res = claude.resolve_desktop_sessions_dir(environ=env, target_home=fake_nonexistent) + assert res is None + finally: + sys.platform = orig_platform + + +def test_find_desktop_project_dir_matching_cwd(tmp_path: Path) -> None: + sessions_dir = tmp_path / "claude-code-sessions" + org1_proj1 = sessions_dir / "org1" / "proj1" + org1_proj1.mkdir(parents=True) + org2_proj2 = sessions_dir / "org2" / "proj2" + org2_proj2.mkdir(parents=True) + + target_cwd = (tmp_path / "my_project").resolve() + target_cwd.mkdir(parents=True) + + pointer1 = org1_proj1 / "local_111.json" + pointer1.write_text(json.dumps({"cwd": str(target_cwd)})) + + pointer2 = org2_proj2 / "local_222.json" + pointer2.write_text(json.dumps({"cwd": "/some/other/repo"})) + + selected = claude.find_desktop_project_dir(sessions_dir, target_cwd) + assert selected == org1_proj1 + + +def test_find_desktop_project_dir_fallback_mtime(tmp_path: Path) -> None: + sessions_dir = tmp_path / "claude-code-sessions" + org1_proj1 = sessions_dir / "org1" / "proj1" + org1_proj1.mkdir(parents=True) + org2_proj2 = sessions_dir / "org2" / "proj2" + org2_proj2.mkdir(parents=True) + + p1 = org1_proj1 / "local_111.json" + p1.write_text(json.dumps({"cwd": "/old/repo"})) + os.utime(p1, (time.time() - 100, time.time() - 100)) + + p2 = org2_proj2 / "local_222.json" + p2.write_text(json.dumps({"cwd": "/recent/repo"})) + os.utime(p2, (time.time(), time.time())) + + # Looking for a cwd that doesn't match any existing pointer + unmatched_cwd = tmp_path / "unmatched" + selected = claude.find_desktop_project_dir(sessions_dir, unmatched_cwd) + assert selected == org2_proj2 + + +def test_find_desktop_project_dir_empty_initializes_defaults(tmp_path: Path) -> None: + sessions_dir = tmp_path / "claude-code-sessions" + sessions_dir.mkdir() + + selected = claude.find_desktop_project_dir(sessions_dir, tmp_path / "any") + expected = ( + sessions_dir + / claude.DEFAULT_DESKTOP_ORG_ID + / claude.DEFAULT_DESKTOP_PROJECT_ID + ) + assert selected == expected + + +def test_desktop_pointer_data_structure(tmp_path: Path) -> None: + session = _sample_session(tmp_path) + data = claude.desktop_pointer_data( + session_id=session.session_id, + cwd=session.cwd, + timestamp=session.started_at, + model=session.model, + title=session.title, + events=session.events, + ) + + assert data["sessionId"] == f"local_{TEST_SESSION_ID}" + assert data["cliSessionId"] == TEST_SESSION_ID + assert data["cwd"] == str(session.cwd.resolve()) + assert data["originCwd"] == str(session.cwd.resolve()) + assert data["model"] == "claude-3-5-sonnet-20241022" + assert data["title"] == "Test Desktop Session" + assert data["isArchived"] is False + assert data["titleSource"] == "custom" + assert data["permissionMode"] == "default" + assert data["importedFrom"] == "local-1p-code" + assert data["alwaysAllowedReasons"] == [] + assert data["sessionPermissionUpdates"] == [] + + # Timestamp in ms: 2026-08-17T12:00:00Z = 1786968000000 + assert isinstance(data["createdAt"], int) + assert isinstance(data["lastActivityAt"], int) + # The second event is at 12:05:00Z, so lastActivityAt should be greater than createdAt + assert data["lastActivityAt"] > data["createdAt"] + assert data["lastFocusedAt"] == data["lastActivityAt"] + + +def test_install_claude_artifact_with_desktop_pointer(tmp_path: Path) -> None: + session = _sample_session(tmp_path) + options = ConversionOptions( + target_format=TargetFormat.CLAUDE, + session_id=TEST_SESSION_ID, + cwd=session.cwd, + ) + artifact = convert_session(session, options) + + target_home = tmp_path / "claude_home" + desktop_dir = tmp_path / "claude_desktop" / "claude-code-sessions" + desktop_dir.mkdir(parents=True) + + native_path, manifest_path, pointer_path = install_claude_artifact( + artifact, + target_home=target_home, + desktop_dir=desktop_dir, + ) + + assert native_path.is_file() + assert manifest_path.is_file() + assert pointer_path is not None + assert pointer_path.is_file() + assert pointer_path.name == f"local_{TEST_SESSION_ID}.json" + + # Verify pointer contents + data = json.loads(pointer_path.read_text(encoding="utf-8")) + assert data["cliSessionId"] == TEST_SESSION_ID + assert data["sessionId"] == f"local_{TEST_SESSION_ID}" + assert data["cwd"] == str(session.cwd.resolve()) + + +def test_install_claude_artifact_dry_run(tmp_path: Path) -> None: + session = _sample_session(tmp_path) + options = ConversionOptions( + target_format=TargetFormat.CLAUDE, + session_id=TEST_SESSION_ID, + cwd=session.cwd, + ) + artifact = convert_session(session, options) + + target_home = tmp_path / "claude_home" + desktop_dir = tmp_path / "claude_desktop" / "claude-code-sessions" + desktop_dir.mkdir(parents=True) + + native_path, manifest_path, pointer_path = install_claude_artifact( + artifact, + target_home=target_home, + desktop_dir=desktop_dir, + dry_run=True, + ) + + assert not native_path.exists() + assert not manifest_path.exists() + assert pointer_path is not None + assert not pointer_path.exists() + + +def test_install_claude_artifact_collision_fails(tmp_path: Path) -> None: + session = _sample_session(tmp_path) + options = ConversionOptions( + target_format=TargetFormat.CLAUDE, + session_id=TEST_SESSION_ID, + cwd=session.cwd, + ) + artifact = convert_session(session, options) + + target_home = tmp_path / "claude_home" + desktop_dir = tmp_path / "claude_desktop" / "claude-code-sessions" + desktop_dir.mkdir(parents=True) + + # First install succeeds + install_claude_artifact(artifact, target_home=target_home, desktop_dir=desktop_dir) + + # Second install should raise error because files exist + with pytest.raises(SessionMigrateError, match="refusing to overwrite"): + install_claude_artifact(artifact, target_home=target_home, desktop_dir=desktop_dir) + + +def test_write_claude_artifact_with_desktop_dir(tmp_path: Path) -> None: + session = _sample_session(tmp_path) + options = ConversionOptions( + target_format=TargetFormat.CLAUDE, + session_id=TEST_SESSION_ID, + cwd=session.cwd, + ) + artifact = convert_session(session, options) + + output_path = tmp_path / "out.jsonl" + manifest_path = tmp_path / "manifest.json" + desktop_dir = tmp_path / "claude_desktop" / "claude-code-sessions" + desktop_dir.mkdir(parents=True) + + out_p, man_p, point_p = write_claude_artifact( + artifact, + output_path=output_path, + manifest_path=manifest_path, + desktop_dir=desktop_dir, + ) + + assert out_p.is_file() + assert man_p.is_file() + assert point_p is not None + assert point_p.is_file() + data = json.loads(point_p.read_text(encoding="utf-8")) + assert data["cliSessionId"] == TEST_SESSION_ID + + +def test_cli_import_to_claude_with_desktop_flag(tmp_path: Path) -> None: + claude_home = tmp_path / "target_claude" + desktop_dir = tmp_path / "desktop" / "claude-code-sessions" + desktop_dir.mkdir(parents=True) + target_sid = "11111111-2222-3333-4444-555555555555" + + code = main( + [ + "import", + str(SOURCE_CLAUDE_FILE), + "--to", + "claude", + "--home", + str(claude_home), + "--session-id", + target_sid, + "--claude-desktop-dir", + str(desktop_dir), + ] + ) + assert code == 0 + + pointer = ( + desktop_dir + / claude.DEFAULT_DESKTOP_ORG_ID + / claude.DEFAULT_DESKTOP_PROJECT_ID + / f"local_{target_sid}.json" + ) + assert pointer.is_file() + data = json.loads(pointer.read_text(encoding="utf-8")) + assert data["cliSessionId"] == target_sid diff --git a/tests/test_target_integration.py b/tests/test_target_integration.py index 251d599..6c6dce1 100644 --- a/tests/test_target_integration.py +++ b/tests/test_target_integration.py @@ -143,6 +143,7 @@ def test_source_and_target_enums_are_deliberately_separate() -> None: AgentFormat.HERMES, AgentFormat.MASTRACODE, AgentFormat.DEVIN, + AgentFormat.CLAUDE_CLOUD, ) assert set(TargetFormat) == { TargetFormat.CLAUDE,