from __future__ import annotations import filecmp import json import os import re import shutil import tempfile from pathlib import Path from runtime_paths import runtime_dir MERGEABLE_JSONL_NAMES = {"history.jsonl", "session_index.jsonl"} DISPOSABLE_STATE_DIR_NAMES = {".tmp", "cache", "tmp"} SHARED_SINGLETON_NAMES = {"installation_id"} EPHEMERAL_RUNTIME_SUFFIXES = (".lock", ".pid", ".sock", ".socket") class SharedCodexState: def __init__( self, legacy_codex_home: Path | None = None, accounts_dir: Path | None = None, ) -> None: self.legacy_codex_home = ( Path(legacy_codex_home) if legacy_codex_home is not None else Path.home() / ".codex" ) self.accounts_dir = ( Path(accounts_dir) if accounts_dir is not None else runtime_dir() / "accounts" ) def sync_shared_links(self, account_home: Path) -> list[Path]: account_home = Path(account_home) account_home.mkdir(parents=True, exist_ok=True) self.legacy_codex_home.mkdir(parents=True, exist_ok=True) names = {path.name for path in account_home.iterdir()} names.update(path.name for path in self.legacy_codex_home.iterdir()) conflicts: list[Path] = [] for name in sorted(names): if ( self._is_credential_name(name) or self._is_sqlite_state_name(name) or self._is_ephemeral_runtime_name(name) ): self._unshare_account_entry(account_home / name) continue conflicts.extend( self._merge_and_link_shared_entry( account_home / name, self.legacy_codex_home / name, ) ) return conflicts def sync_all_shared_links(self) -> list[Path]: if not self.accounts_dir.exists(): return [] conflicts: list[Path] = [] for account_dir in sorted(self.accounts_dir.iterdir()): codex_home = account_dir / "CODEX_HOME" if codex_home.is_dir(): conflicts.extend(self.sync_shared_links(codex_home)) return conflicts def _unshare_account_entry(self, entry: Path) -> None: if not entry.is_symlink(): return target = entry.resolve() try: target.relative_to(self.legacy_codex_home.resolve()) except ValueError: return if target.is_dir(): return entry.unlink() if target.is_file(): shutil.copy2(target, entry) def _merge_and_link_shared_entry(self, source: Path, destination: Path) -> list[Path]: if source.is_symlink(): if source.resolve() == destination.resolve(): return [] self._unshare_foreign_link(source) return [source] if source.exists(): if not destination.exists() and not destination.is_symlink(): os.replace(source, destination) elif source.is_dir() and destination.is_dir(): conflicts = self._merge_directory( source, destination, discard_conflicts=source.name in DISPOSABLE_STATE_DIR_NAMES, ) if conflicts: return conflicts source.rmdir() elif source.is_file() and destination.is_file(): if not self._merge_file(source, destination): return [source] else: return [source] if destination.exists(): source.symlink_to(destination, target_is_directory=destination.is_dir()) return [] def _merge_directory( self, source: Path, destination: Path, discard_conflicts: bool = False, ) -> list[Path]: destination.mkdir(parents=True, exist_ok=True) conflicts: list[Path] = [] for item in sorted(source.iterdir()): target = destination / item.name if item.is_dir() and not item.is_symlink(): if target.exists() and not target.is_dir(): conflicts.append(item) continue nested = self._merge_directory(item, target, discard_conflicts) if nested: conflicts.extend(nested) continue item.rmdir() continue if not target.exists() and not target.is_symlink(): os.replace(item, target) continue if item.is_file() and target.is_file(): if not self._merge_file(item, target, discard_conflicts): conflicts.append(item) continue conflicts.append(item) return conflicts def _merge_file( self, source: Path, destination: Path, discard_conflicts: bool = False, ) -> bool: if filecmp.cmp(source, destination, shallow=False): source.unlink() return True if source.name in MERGEABLE_JSONL_NAMES: self._merge_jsonl(source, destination) return True if source.suffix == ".jsonl" and self._resolve_transcript(source, destination): return True if source.suffix == ".log": self._append_file(source, destination) return True if discard_conflicts or source.name in SHARED_SINGLETON_NAMES: source.unlink() return True return False # A session transcript is append-only and named by session id, so two copies of one id are # the same session recorded twice and the shorter is a byte-prefix of the longer. Calling # that a conflict left the whole projects/ tree unshared, and every session recorded under # one account then vanished from `--resume` under any other. @staticmethod def _resolve_transcript(source: Path, destination: Path) -> bool: shorter, longer = sorted((source, destination), key=lambda p: p.stat().st_size) with longer.open("rb") as handle: head = handle.read(shorter.stat().st_size) if head != shorter.read_bytes(): return False if shorter == destination: os.replace(source, destination) else: source.unlink() return True def _unshare_foreign_link(self, source: Path) -> None: target = source.resolve() source.unlink() if target.is_file(): shutil.copy2(target, source) @staticmethod def _append_file(source: Path, destination: Path) -> None: source_bytes = source.read_bytes() needs_separator = False if destination.stat().st_size: with destination.open("rb") as handle: handle.seek(-1, os.SEEK_END) needs_separator = handle.read(1) != b"\n" with destination.open("ab") as handle: if needs_separator and source_bytes: handle.write(b"\n") handle.write(source_bytes) source.unlink() def _merge_jsonl(self, source: Path, destination: Path) -> None: lines = destination.read_text(encoding="utf-8").splitlines() lines.extend(source.read_text(encoding="utf-8").splitlines()) lines = list(dict.fromkeys(line for line in lines if line)) if destination.name == "history.jsonl": lines.sort(key=self._history_sort_key) else: lines.sort(key=self._session_index_sort_key) self._atomic_write_text(destination, "\n".join(lines) + ("\n" if lines else "")) source.unlink() @staticmethod def _history_sort_key(line: str) -> tuple[bool, int]: try: timestamp = json.loads(line).get("ts") except (json.JSONDecodeError, AttributeError): timestamp = None return (not isinstance(timestamp, int), timestamp if isinstance(timestamp, int) else 0) @staticmethod def _session_index_sort_key(line: str) -> tuple[bool, str]: try: timestamp = json.loads(line).get("updated_at") except (json.JSONDecodeError, AttributeError): timestamp = None return (not isinstance(timestamp, str), timestamp if isinstance(timestamp, str) else "") @staticmethod def _is_credential_name(name: str) -> bool: return name == "auth.json" or name.startswith("auth.") @staticmethod def _is_ephemeral_runtime_name(name: str) -> bool: return name.endswith(EPHEMERAL_RUNTIME_SUFFIXES) @staticmethod def _is_sqlite_state_name(name: str) -> bool: return re.search(r"\.sqlite(?:-(?:wal|shm|journal))?$", name) is not None @staticmethod def _atomic_write_text(path: Path, content: str) -> None: path.parent.mkdir(parents=True, exist_ok=True) with tempfile.NamedTemporaryFile( "w", encoding="utf-8", dir=path.parent, delete=False, ) as handle: handle.write(content) temp_name = handle.name os.replace(temp_name, path)