feat(claude): support Claude Desktop import pointers and Claude Cloud migration

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)

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"

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,
     *,

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))

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",